Skip to content

Commit 102a416

Browse files
owaiskazi19owaiskazi19
committed
Addressed PR Comments
Signed-off-by: Owais Kazi <[email protected]>
1 parent ba7e1d4 commit 102a416

File tree

6 files changed

+30
-27
lines changed

6 files changed

+30
-27
lines changed

‎src/main/java/org/opensearch/sdk/ClusterSettingResponseHandler.java

+1-2
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,7 @@ public void handleResponse(ClusterSettingsResponse response) {
2424

2525
@Override
2626
public void handleException(TransportException exp) {
27-
logger.debug(new ParameterizedMessage("ClusterSettingRequest failed"), exp);
28-
// inProgressLatch.countDown();
27+
logger.info("ClusterSettingRequest failed", exp);
2928
}
3029

3130
@Override

‎src/main/java/org/opensearch/sdk/ClusterStateResponseHandler.java renamed to ‎src/main/java/org/opensearch/sdk/ExtensionClusterStateResponseHandler.java

+3-4
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,8 @@
1212
import java.io.IOException;
1313
import java.util.concurrent.CountDownLatch;
1414

15-
public class ClusterStateResponseHandler implements TransportResponseHandler<ExtensionClusterStateResponse> {
16-
private static final Logger logger = LogManager.getLogger(ClusterStateResponseHandler.class);
15+
public class ExtensionClusterStateResponseHandler implements TransportResponseHandler<ExtensionClusterStateResponse> {
16+
private static final Logger logger = LogManager.getLogger(ExtensionClusterStateResponseHandler.class);
1717
final CountDownLatch inProgressLatch = new CountDownLatch(1);
1818

1919
@Override
@@ -24,8 +24,7 @@ public void handleResponse(ExtensionClusterStateResponse response) {
2424

2525
@Override
2626
public void handleException(TransportException exp) {
27-
logger.debug(new ParameterizedMessage("ClusterStateRequest failed"), exp);
28-
// inProgressLatch.countDown();
27+
logger.info("ClusterStateRequest failed", exp);
2928
}
3029

3130
@Override

‎src/main/java/org/opensearch/sdk/ExtensionsRunner.java

+21-15
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import org.opensearch.cluster.ClusterModule;
2020
import org.opensearch.cluster.ExtensionRequest;
2121
import org.opensearch.cluster.node.DiscoveryNode;
22+
import org.opensearch.common.concurrent.CompletableContext;
2223
import org.opensearch.common.io.stream.NamedWriteableRegistry;
2324
import org.opensearch.common.network.NetworkModule;
2425
import org.opensearch.common.network.NetworkService;
@@ -49,6 +50,7 @@
4950
import java.io.IOException;
5051
import java.util.Collections;
5152
import java.util.List;
53+
import java.util.concurrent.CompletableFuture;
5254
import java.util.concurrent.CountDownLatch;
5355
import java.util.concurrent.TimeUnit;
5456
import java.util.function.Function;
@@ -59,12 +61,9 @@
5961
import static org.opensearch.common.UUIDs.randomBase64UUID;
6062

6163
public class ExtensionsRunner {
62-
63-
public static final String REQUEST_EXTENSION_ACTION_NAME = "internal:discovery/extensions";
64-
6564
private static ExtensionSettings extensionSettings = null;
66-
6765
private DiscoveryNode opensearchNode = null;
66+
TransportService transportService = null;
6867

6968
static {
7069
try {
@@ -93,35 +92,37 @@ public static ExtensionSettings getExtensionSettings() throws IOException {
9392
}
9493

9594
PluginResponse handlePluginsRequest(PluginRequest pluginRequest) {
96-
logger.info("Handling Plugins Request");
95+
logger.info("Registering Plugin Request received from OpenSearch");
9796
PluginResponse pluginResponse = new PluginResponse("RealExtension");
9897
opensearchNode = pluginRequest.getSourceNode();
9998
return pluginResponse;
10099
}
101100

102-
IndicesModuleResponse handleIndicesModuleRequest(IndicesModuleRequest indicesModuleRequest, TransportService transportService) {
103-
logger.info("Indices Module Request");
101+
IndicesModuleResponse handleIndicesModuleRequest(IndicesModuleRequest indicesModuleRequest) {
102+
logger.info("Registering Indices Module Request received from OpenSearch");
104103
IndicesModuleResponse indicesModuleResponse = new IndicesModuleResponse(true, true, true);
105104

106105
// CreateComponent
107106
transportService.connectToNode(opensearchNode);
108107
final CountDownLatch inProgressLatch = new CountDownLatch(1);
109108
try {
110-
logger.info("Sending request to opensearch");
111-
ClusterStateResponseHandler clusterStateResponseHandler = new ClusterStateResponseHandler();
109+
logger.info("Sending Cluster State request to OpenSearch after creating index");
110+
ExtensionClusterStateResponseHandler clusterStateResponseHandler = new ExtensionClusterStateResponseHandler();
112111
transportService.sendRequest(
113112
opensearchNode,
114113
ExtensionsOrchestrator.REQUEST_EXTENSION_CLUSTER_STATE,
115114
new ExtensionRequest(ExtensionsOrchestrator.RequestType.REQUEST_EXTENSION_CLUSTER_STATE),
116115
clusterStateResponseHandler
117116
);
117+
logger.info("Sending Cluster Settings request to OpenSearch after creating index");
118118
ClusterSettingResponseHandler clusterSettingResponseHandler = new ClusterSettingResponseHandler();
119119
transportService.sendRequest(
120120
opensearchNode,
121121
ExtensionsOrchestrator.REQUEST_EXTENSION_CLUSTER_SETTINGS,
122122
new ExtensionRequest(ExtensionsOrchestrator.RequestType.REQUEST_EXTENSION_CLUSTER_SETTINGS),
123123
clusterSettingResponseHandler
124124
);
125+
logger.info("Sending Local Node request to OpenSearch after creating index");
125126
LocalNodeResponseHandler localNodeResponseHandler = new LocalNodeResponseHandler();
126127
transportService.sendRequest(
127128
opensearchNode,
@@ -130,7 +131,7 @@ IndicesModuleResponse handleIndicesModuleRequest(IndicesModuleRequest indicesMod
130131
localNodeResponseHandler
131132
);
132133
inProgressLatch.await(1, TimeUnit.SECONDS);
133-
logger.info("Received response from OpenSearch");
134+
logger.info("Received response from OpenSearch for ClusterState, ClusterSettings and LocalNode");
134135
} catch (Exception e) {
135136
e.printStackTrace();
136137
logger.error(e.toString());
@@ -141,7 +142,7 @@ IndicesModuleResponse handleIndicesModuleRequest(IndicesModuleRequest indicesMod
141142

142143
// Works as beforeIndexRemoved
143144
IndicesModuleNameResponse handleIndicesModuleNameRequest(IndicesModuleRequest indicesModuleRequest) {
144-
logger.info("Indices Module Name Request");
145+
logger.info("Registering Indices Module Name Request received from OpenSearch");
145146
IndicesModuleNameResponse indicesModuleNameResponse = new IndicesModuleNameResponse(true);
146147
return indicesModuleNameResponse;
147148
}
@@ -180,7 +181,7 @@ public Netty4Transport getNetty4Transport(Settings settings, ThreadPool threadPo
180181
return transport;
181182
}
182183

183-
public TransportService getTransportService(Settings settings) throws IOException {
184+
public TransportService createTransportService(Settings settings) throws IOException {
184185

185186
ThreadPool threadPool = new ThreadPool(settings);
186187

@@ -214,7 +215,7 @@ public void startTransportService(TransportService transportService) {
214215
transportService.start();
215216
transportService.acceptIncomingRequests();
216217
transportService.registerRequestHandler(
217-
REQUEST_EXTENSION_ACTION_NAME,
218+
ExtensionsOrchestrator.REQUEST_EXTENSION_ACTION_NAME,
218219
ThreadPool.Names.GENERIC,
219220
false,
220221
false,
@@ -228,7 +229,7 @@ public void startTransportService(TransportService transportService) {
228229
false,
229230
false,
230231
IndicesModuleRequest::new,
231-
((request, channel, task) -> channel.sendResponse(handleIndicesModuleRequest(request, transportService)))
232+
((request, channel, task) -> channel.sendResponse(handleIndicesModuleRequest(request)))
232233

233234
);
234235
transportService.registerRequestHandler(
@@ -242,6 +243,10 @@ public void startTransportService(TransportService transportService) {
242243

243244
}
244245

246+
public void setTransportService(TransportService transportService) {
247+
this.transportService = transportService;
248+
}
249+
245250
// manager method for action listener
246251
public void startActionListener(int timeout) {
247252
final ActionListener actionListener = new ActionListener();
@@ -253,7 +258,8 @@ public static void main(String[] args) throws IOException {
253258
ExtensionsRunner extensionsRunner = new ExtensionsRunner();
254259

255260
// configure and retrieve transport service with settings
256-
TransportService transportService = extensionsRunner.getTransportService(settings);
261+
TransportService transportService = extensionsRunner.createTransportService(settings);
262+
extensionsRunner.setTransportService(transportService);
257263

258264
// start transport service and action listener
259265
extensionsRunner.startTransportService(transportService);

‎src/main/java/org/opensearch/sdk/LocalNodeResponseHandler.java

+1-2
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,7 @@ public void handleResponse(LocalNodeResponse response) {
2424

2525
@Override
2626
public void handleException(TransportException exp) {
27-
logger.debug(new ParameterizedMessage("LocalNodeRequest failed"), exp);
28-
// inProgressLatch.countDown();
27+
logger.info("LocalNodeRequest failed", exp);
2928
}
3029

3130
@Override

‎src/test/java/org/opensearch/sdk/TestMainScript.java

+3-3
Original file line numberDiff line numberDiff line change
@@ -36,15 +36,15 @@ public void setUp() throws IOException {
3636
// test ExtensionsRunner getTransportService return type is transport service
3737
@Test
3838
public void testGetTransportService() throws IOException {
39-
assert (extensionsRunner.getTransportService(settings) instanceof TransportService);
39+
assert (extensionsRunner.createTransportService(settings) instanceof TransportService);
4040
}
4141

4242
// test manager method invokes start on transport service
4343
@Test
4444
public void testTransportServiceStarted() throws IOException {
4545

4646
// retrieve and mock transport service
47-
TransportService transportService = Mockito.spy(extensionsRunner.getTransportService(settings));
47+
TransportService transportService = Mockito.spy(extensionsRunner.createTransportService(settings));
4848

4949
// verify mocked object interaction in manager method
5050
extensionsRunner.startTransportService(transportService);
@@ -56,7 +56,7 @@ public void testTransportServiceStarted() throws IOException {
5656
public void testTransportServiceAcceptedIncomingRequests() throws IOException {
5757

5858
// retrieve and mock transport service
59-
TransportService transportService = Mockito.spy(extensionsRunner.getTransportService(settings));
59+
TransportService transportService = Mockito.spy(extensionsRunner.createTransportService(settings));
6060

6161
// verify mocked object interaction in manager method
6262
extensionsRunner.startTransportService(transportService);

‎src/test/java/org/opensearch/sdk/TransportCommunicationIT.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ private void startTransportandClient(Settings settings, Thread client) throws IO
144144

145145
// retrieve transport service
146146
ExtensionsRunner extensionsRunner = new ExtensionsRunner();
147-
TransportService transportService = extensionsRunner.getTransportService(settings);
147+
TransportService transportService = extensionsRunner.createTransportService(settings);
148148

149149
// start transport service
150150
extensionsRunner.startTransportService(transportService);

0 commit comments

Comments
 (0)