Skip to content

Commit 4e3d7ee

Browse files
jasontedorjasontedor
committed
Invoke response handler on failure to send (#53631)
Today it can happen that a transport message fails to send (for example, because a transport interceptor rejects the request). In this case, the response handler is never invoked, which can lead to necessary cleanups not being performed. There are two ways to handle this. One is to expect every callsite that sends a message to try/catch these exceptions and handle them appropriately. The other is merely to invoke the response handler to handle the exception, which is already equipped to handle transport exceptions.
1 parent 01af65d commit 4e3d7ee

File tree

6 files changed

+169
-36
lines changed

6 files changed

+169
-36
lines changed

‎modules/transport-netty4/src/test/java/org/elasticsearch/transport/netty4/SimpleNetty4TransportTests.java‎

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
import org.elasticsearch.transport.ConnectionProfile;
3939
import org.elasticsearch.transport.TcpChannel;
4040
import org.elasticsearch.transport.Transport;
41+
import org.elasticsearch.transport.TransportInterceptor;
4142
import org.elasticsearch.transport.TransportSettings;
4243

4344
import java.net.InetAddress;
@@ -46,12 +47,14 @@
4647

4748
import static java.util.Collections.emptyMap;
4849
import static java.util.Collections.emptySet;
50+
import static org.elasticsearch.transport.TransportService.NOOP_TRANSPORT_INTERCEPTOR;
4951
import static org.hamcrest.Matchers.containsString;
5052

5153
public class SimpleNetty4TransportTests extends AbstractSimpleTransportTestCase {
5254

5355
public static MockTransportService nettyFromThreadPool(Settings settings, ThreadPool threadPool, final Version version,
54-
ClusterSettings clusterSettings, boolean doHandshake) {
56+
ClusterSettings clusterSettings, boolean doHandshake,
57+
TransportInterceptor interceptor) {
5558
NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry(Collections.emptyList());
5659
Transport transport = new Netty4Transport(settings, version, threadPool, new NetworkService(Collections.emptyList()),
5760
PageCacheRecycler.NON_RECYCLING_INSTANCE, namedWriteableRegistry, new NoneCircuitBreakerService()) {
@@ -66,16 +69,28 @@ public void executeHandshake(DiscoveryNode node, TcpChannel channel, ConnectionP
6669
}
6770
}
6871
};
69-
MockTransportService mockTransportService =
70-
MockTransportService.createNewService(settings, transport, version, threadPool, clusterSettings, Collections.emptySet());
72+
MockTransportService mockTransportService = MockTransportService.createNewService(
73+
settings,
74+
transport,
75+
version,
76+
threadPool,
77+
clusterSettings,
78+
Collections.emptySet(),
79+
interceptor);
7180
mockTransportService.start();
7281
return mockTransportService;
7382
}
7483

7584
@Override
76-
protected MockTransportService build(Settings settings, Version version, ClusterSettings clusterSettings, boolean doHandshake) {
85+
protected MockTransportService build(
86+
Settings settings,
87+
Version version,
88+
ClusterSettings clusterSettings,
89+
boolean doHandshake,
90+
TransportInterceptor interceptor) {
7791
settings = Settings.builder().put(settings).put(TransportSettings.PORT.getKey(), "0").build();
78-
MockTransportService transportService = nettyFromThreadPool(settings, threadPool, version, clusterSettings, doHandshake);
92+
MockTransportService transportService =
93+
nettyFromThreadPool(settings, threadPool, version, clusterSettings, doHandshake, interceptor);
7994
transportService.start();
8095
return transportService;
8196
}
@@ -103,7 +118,7 @@ public void testBindUnavailableAddress() {
103118
ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
104119
BindTransportException bindTransportException = expectThrows(BindTransportException.class, () -> {
105120
MockTransportService transportService =
106-
nettyFromThreadPool(settings, threadPool, Version.CURRENT, clusterSettings, true);
121+
nettyFromThreadPool(settings, threadPool, Version.CURRENT, clusterSettings, true, NOOP_TRANSPORT_INTERCEPTOR);
107122
try {
108123
transportService.start();
109124
} finally {

‎server/src/main/java/org/elasticsearch/transport/TransportService.java‎

Lines changed: 34 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,6 @@
4848
import org.elasticsearch.common.util.concurrent.ThreadContext;
4949
import org.elasticsearch.core.internal.io.IOUtils;
5050
import org.elasticsearch.tasks.Task;
51-
import org.elasticsearch.tasks.TaskCancelledException;
5251
import org.elasticsearch.tasks.TaskManager;
5352
import org.elasticsearch.threadpool.Scheduler;
5453
import org.elasticsearch.threadpool.ThreadPool;
@@ -512,37 +511,57 @@ public <T extends TransportResponse> TransportFuture<T> submitRequest(DiscoveryN
512511
public <T extends TransportResponse> void sendRequest(final DiscoveryNode node, final String action,
513512
final TransportRequest request,
514513
final TransportResponseHandler<T> handler) {
514+
final Transport.Connection connection;
515515
try {
516-
Transport.Connection connection = getConnection(node);
517-
sendRequest(connection, action, request, TransportRequestOptions.EMPTY, handler);
518-
} catch (NodeNotConnectedException ex) {
516+
connection = getConnection(node);
517+
} catch (final NodeNotConnectedException ex) {
519518
// the caller might not handle this so we invoke the handler
520519
handler.handleException(ex);
520+
return;
521521
}
522+
sendRequest(connection, action, request, TransportRequestOptions.EMPTY, handler);
522523
}
523524

524525
public final <T extends TransportResponse> void sendRequest(final DiscoveryNode node, final String action,
525526
final TransportRequest request,
526527
final TransportRequestOptions options,
527528
TransportResponseHandler<T> handler) {
529+
final Transport.Connection connection;
528530
try {
529-
Transport.Connection connection = getConnection(node);
530-
sendRequest(connection, action, request, options, handler);
531-
} catch (NodeNotConnectedException ex) {
531+
connection = getConnection(node);
532+
} catch (final NodeNotConnectedException ex) {
532533
// the caller might not handle this so we invoke the handler
533534
handler.handleException(ex);
535+
return;
534536
}
537+
sendRequest(connection, action, request, options, handler);
535538
}
536539

540+
/**
541+
* Sends a request on the specified connection. If there is a failure sending the request, the specified handler is invoked.
542+
*
543+
* @param connection the connection to send the request on
544+
* @param action the name of the action
545+
* @param request the request
546+
* @param options the options for this request
547+
* @param handler the response handler
548+
* @param <T> the type of the transport response
549+
*/
537550
public final <T extends TransportResponse> void sendRequest(final Transport.Connection connection, final String action,
538551
final TransportRequest request,
539552
final TransportRequestOptions options,
540553
TransportResponseHandler<T> handler) {
541554
try {
542555
asyncSender.sendRequest(connection, action, request, options, handler);
543-
} catch (NodeNotConnectedException ex) {
556+
} catch (final Exception ex) {
544557
// the caller might not handle this so we invoke the handler
545-
handler.handleException(ex);
558+
final TransportException te;
559+
if (ex instanceof TransportException) {
560+
te = (TransportException) ex;
561+
} else {
562+
te = new TransportException("failure to send", ex);
563+
}
564+
handler.handleException(te);
546565
}
547566
}
548567

@@ -562,13 +581,15 @@ public final <T extends TransportResponse> void sendChildRequest(final Discovery
562581
final TransportRequest request, final Task parentTask,
563582
final TransportRequestOptions options,
564583
final TransportResponseHandler<T> handler) {
584+
final Transport.Connection connection;
565585
try {
566-
Transport.Connection connection = getConnection(node);
567-
sendChildRequest(connection, action, request, parentTask, options, handler);
568-
} catch (NodeNotConnectedException ex) {
586+
connection = getConnection(node);
587+
} catch (final NodeNotConnectedException ex) {
569588
// the caller might not handle this so we invoke the handler
570589
handler.handleException(ex);
590+
return;
571591
}
592+
sendChildRequest(connection, action, request, parentTask, options, handler);
572593
}
573594

574595
public <T extends TransportResponse> void sendChildRequest(final Transport.Connection connection, final String action,
@@ -582,16 +603,7 @@ public <T extends TransportResponse> void sendChildRequest(final Transport.Conne
582603
final TransportRequestOptions options,
583604
final TransportResponseHandler<T> handler) {
584605
request.setParentTask(localNode.getId(), parentTask.getId());
585-
try {
586-
sendRequest(connection, action, request, options, handler);
587-
} catch (TaskCancelledException ex) {
588-
// The parent task is already cancelled - just fail the request
589-
handler.handleException(new TransportException(ex));
590-
} catch (NodeNotConnectedException ex) {
591-
// the caller might not handle this so we invoke the handler
592-
handler.handleException(ex);
593-
}
594-
606+
sendRequest(connection, action, request, options, handler);
595607
}
596608

597609
private <T extends TransportResponse> void sendRequestInternal(final Transport.Connection connection, final String action,

‎test/framework/src/main/java/org/elasticsearch/test/transport/MockTransportService.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,13 @@ public static MockTcpTransport newMockTransport(Settings settings, Version versi
117117

118118
public static MockTransportService createNewService(Settings settings, Transport transport, Version version, ThreadPool threadPool,
119119
@Nullable ClusterSettings clusterSettings, Set<String> taskHeaders) {
120-
return new MockTransportService(settings, transport, threadPool, TransportService.NOOP_TRANSPORT_INTERCEPTOR,
120+
return createNewService(settings, transport, version, threadPool, clusterSettings, taskHeaders, NOOP_TRANSPORT_INTERCEPTOR);
121+
}
122+
123+
public static MockTransportService createNewService(Settings settings, Transport transport, Version version, ThreadPool threadPool,
124+
@Nullable ClusterSettings clusterSettings, Set<String> taskHeaders,
125+
TransportInterceptor interceptor) {
126+
return new MockTransportService(settings, transport, threadPool, interceptor,
121127
boundAddress ->
122128
new DiscoveryNode(Node.NODE_NAME_SETTING.get(settings), UUIDs.randomBase64UUID(), boundAddress.publishAddress(),
123129
Node.NODE_ATTRIBUTES.getAsMap(settings), DiscoveryNode.getRolesFromSettings(settings), version),

‎test/framework/src/main/java/org/elasticsearch/transport/AbstractSimpleTransportTestCase.java‎

Lines changed: 95 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -87,12 +87,15 @@
8787

8888
import static java.util.Collections.emptyMap;
8989
import static java.util.Collections.emptySet;
90+
import static org.elasticsearch.transport.TransportService.NOOP_TRANSPORT_INTERCEPTOR;
9091
import static org.hamcrest.Matchers.containsString;
9192
import static org.hamcrest.Matchers.empty;
9293
import static org.hamcrest.Matchers.equalTo;
9394
import static org.hamcrest.Matchers.hasToString;
9495
import static org.hamcrest.Matchers.instanceOf;
96+
import static org.hamcrest.Matchers.not;
9597
import static org.hamcrest.Matchers.notNullValue;
98+
import static org.hamcrest.Matchers.nullValue;
9699
import static org.hamcrest.Matchers.startsWith;
97100

98101
public abstract class AbstractSimpleTransportTestCase extends ESTestCase {
@@ -111,7 +114,16 @@ public abstract class AbstractSimpleTransportTestCase extends ESTestCase {
111114
protected volatile DiscoveryNode nodeB;
112115
protected volatile MockTransportService serviceB;
113116

114-
protected abstract MockTransportService build(Settings settings, Version version, ClusterSettings clusterSettings, boolean doHandshake);
117+
private MockTransportService build(Settings settings, Version version, ClusterSettings clusterSettings, boolean doHandshake) {
118+
return build(settings, version, clusterSettings, doHandshake, NOOP_TRANSPORT_INTERCEPTOR);
119+
}
120+
121+
protected abstract MockTransportService build(
122+
Settings settings,
123+
Version version,
124+
ClusterSettings clusterSettings,
125+
boolean doHandshake,
126+
TransportInterceptor interceptor);
115127

116128
protected int channelsPerNodeConnection() {
117129
// This is a customized profile for this test case.
@@ -165,6 +177,12 @@ public void onNodeDisconnected(DiscoveryNode node) {
165177

166178
private MockTransportService buildService(final String name, final Version version, ClusterSettings clusterSettings,
167179
Settings settings, boolean acceptRequests, boolean doHandshake) {
180+
return buildService(name, version, clusterSettings, settings, acceptRequests, doHandshake, NOOP_TRANSPORT_INTERCEPTOR);
181+
}
182+
183+
private MockTransportService buildService(final String name, final Version version, ClusterSettings clusterSettings,
184+
Settings settings, boolean acceptRequests, boolean doHandshake,
185+
TransportInterceptor interceptor) {
168186
MockTransportService service = build(
169187
Settings.builder()
170188
.put(settings)
@@ -173,7 +191,7 @@ private MockTransportService buildService(final String name, final Version versi
173191
.put(TransportSettings.TRACE_LOG_EXCLUDE_SETTING.getKey(), "NOTHING")
174192
.build(),
175193
version,
176-
clusterSettings, doHandshake);
194+
clusterSettings, doHandshake, interceptor);
177195
if (acceptRequests) {
178196
service.acceptIncomingRequests();
179197
}
@@ -186,7 +204,7 @@ protected MockTransportService buildService(final String name, final Version ver
186204

187205
protected MockTransportService buildService(final String name, final Version version, ClusterSettings clusterSettings,
188206
Settings settings) {
189-
return buildService(name, version, clusterSettings, settings, true, true);
207+
return buildService(name, version, clusterSettings, settings, true, true, NOOP_TRANSPORT_INTERCEPTOR);
190208
}
191209

192210
@Override
@@ -2731,6 +2749,80 @@ public void onConnectionClosed(Transport.Connection connection) {
27312749
}
27322750
}
27332751

2752+
// test that the response handler is invoked on a failure to send
2753+
public void testFailToSend() throws InterruptedException {
2754+
final RuntimeException failToSendException;
2755+
if (randomBoolean()) {
2756+
failToSendException = new IllegalStateException("fail to send");
2757+
} else {
2758+
failToSendException = new TransportException("fail to send");
2759+
}
2760+
final TransportInterceptor interceptor = new TransportInterceptor() {
2761+
@Override
2762+
public AsyncSender interceptSender(final AsyncSender sender) {
2763+
return new AsyncSender() {
2764+
@Override
2765+
public <T extends TransportResponse> void sendRequest(
2766+
final Transport.Connection connection,
2767+
final String action,
2768+
final TransportRequest request,
2769+
final TransportRequestOptions options,
2770+
final TransportResponseHandler<T> handler) {
2771+
if ("fail-to-send-action".equals(action)) {
2772+
throw failToSendException;
2773+
} else {
2774+
sender.sendRequest(connection, action, request, options, handler);
2775+
}
2776+
}
2777+
};
2778+
}
2779+
};
2780+
try (MockTransportService serviceC = buildService("TS_C", CURRENT_VERSION, null, Settings.EMPTY, true, true, interceptor)) {
2781+
serviceC.start();
2782+
serviceC.acceptIncomingRequests();
2783+
serviceC.connectToNode(serviceA.getLocalDiscoNode(), ConnectionProfile.buildDefaultConnectionProfile(Settings.EMPTY));
2784+
final AtomicReference<TransportException> te = new AtomicReference<>();
2785+
final Transport.Connection connection = serviceC.getConnection(nodeA);
2786+
serviceC.sendRequest(
2787+
connection,
2788+
"fail-to-send-action",
2789+
TransportRequest.Empty.INSTANCE,
2790+
TransportRequestOptions.EMPTY,
2791+
new TransportResponseHandler<TransportResponse>() {
2792+
@Override
2793+
public void handleResponse(final TransportResponse response) {
2794+
fail("handle response should not be invoked");
2795+
}
2796+
2797+
@Override
2798+
public void handleException(final TransportException exp) {
2799+
te.set(exp);
2800+
}
2801+
2802+
@Override
2803+
public String executor() {
2804+
return ThreadPool.Names.SAME;
2805+
}
2806+
2807+
@Override
2808+
public TransportResponse read(final StreamInput in) {
2809+
return TransportResponse.Empty.INSTANCE;
2810+
}
2811+
});
2812+
assertThat(te.get(), not(nullValue()));
2813+
2814+
if (failToSendException instanceof IllegalStateException) {
2815+
assertThat(te.get().getMessage(), equalTo("failure to send"));
2816+
assertThat(te.get().getCause(), instanceOf(IllegalStateException.class));
2817+
assertThat(te.get().getCause().getMessage(), equalTo("fail to send"));
2818+
} else {
2819+
assertThat(te.get().getMessage(), equalTo("fail to send"));
2820+
assertThat(te.get().getCause(), nullValue());
2821+
}
2822+
}
2823+
2824+
}
2825+
27342826
private void closeConnectionChannel(Transport.Connection connection) {
27352827
StubbableTransport.WrappedConnection wrappedConnection = (StubbableTransport.WrappedConnection) connection;
27362828
TcpTransport.NodeChannels channels = (TcpTransport.NodeChannels) wrappedConnection.getConnection();

‎test/framework/src/test/java/org/elasticsearch/transport/MockTcpTransportTests.java‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,8 @@
3434
public class MockTcpTransportTests extends AbstractSimpleTransportTestCase {
3535

3636
@Override
37-
protected MockTransportService build(Settings settings, Version version, ClusterSettings clusterSettings, boolean doHandshake) {
37+
protected MockTransportService build(Settings settings, Version version, ClusterSettings clusterSettings, boolean doHandshake,
38+
TransportInterceptor interceptor) {
3839
NamedWriteableRegistry namedWriteableRegistry = new NamedWriteableRegistry(Collections.emptyList());
3940
Transport transport = new MockTcpTransport(settings, threadPool, BigArrays.NON_RECYCLING_INSTANCE,
4041
new NoneCircuitBreakerService(), namedWriteableRegistry, new NetworkService(Collections.emptyList()), version) {
@@ -49,8 +50,14 @@ public void executeHandshake(DiscoveryNode node, TcpChannel channel, ConnectionP
4950
}
5051
}
5152
};
52-
MockTransportService mockTransportService =
53-
MockTransportService.createNewService(settings, transport, version, threadPool, clusterSettings, Collections.emptySet());
53+
MockTransportService mockTransportService = MockTransportService.createNewService(
54+
settings,
55+
transport,
56+
version,
57+
threadPool,
58+
clusterSettings,
59+
Collections.emptySet(),
60+
interceptor);
5461
mockTransportService.start();
5562
return mockTransportService;
5663
}

0 commit comments

Comments
 (0)