From 2d8b67e2cc6c81d361edfb7195f1772da698e0c1 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Fri, 5 Apr 2019 09:53:19 -0400 Subject: [PATCH 1/7] Suppress lease background sync failures if stopping If the transport service is stopped, likely because we are shutting down, and a retention lease background sync fires the logs will display a warn message and stacktrace. Yet, this situaton is harmless and can happen as a normal course of business when shutting down. This commit suppresses the log messages in this case. --- .../org/elasticsearch/index/IndexService.java | 2 +- .../seqno/RetentionLeaseBackgroundSyncAction.java | 15 +++++++++++++-- .../elasticsearch/transport/TransportService.java | 9 +++++++-- 3 files changed, 21 insertions(+), 5 deletions(-) diff --git a/server/src/main/java/org/elasticsearch/index/IndexService.java b/server/src/main/java/org/elasticsearch/index/IndexService.java index 1a49fd418735d..68966f6d0c946 100644 --- a/server/src/main/java/org/elasticsearch/index/IndexService.java +++ b/server/src/main/java/org/elasticsearch/index/IndexService.java @@ -964,7 +964,7 @@ public String toString() { public static final Setting RETENTION_LEASE_SYNC_INTERVAL_SETTING = Setting.timeSetting( "index.soft_deletes.retention_lease.sync_interval", - new TimeValue(30, TimeUnit.SECONDS), + new TimeValue(25, TimeUnit.MILLISECONDS), new TimeValue(0, TimeUnit.MILLISECONDS), Property.Dynamic, Property.IndexScope); diff --git a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java index d454c2de75b28..a28ec54d0441c 100644 --- a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java +++ b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java @@ -44,6 +44,7 @@ import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.indices.IndicesService; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportService; import java.io.IOException; @@ -113,9 +114,19 @@ public void backgroundSync( ActionListener.wrap( r -> {}, e -> { - if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) == null) { - getLogger().warn(new ParameterizedMessage("{} retention lease background sync failed", shardId), e); + final TransportException maybeTransport = + (TransportException) ExceptionsHelper.unwrap(e, TransportException.class); + if (maybeTransport != null + && (maybeTransport.getMessage().equals("TransportService is closed stopped can't send request") + || maybeTransport.getMessage().equals("transport stopped, action: " + ACTION_NAME + "[p]"))) { + // we are likely shutting down + return; } + if (ExceptionsHelper.unwrap(e, AlreadyClosedException.class, IndexShardClosedException.class) != null) { + // the shard is closed + return; + } + getLogger().warn(new ParameterizedMessage("{} retention lease background sync failed", shardId), e); })); } } diff --git a/server/src/main/java/org/elasticsearch/transport/TransportService.java b/server/src/main/java/org/elasticsearch/transport/TransportService.java index 1288f6fe16f01..54cb7878aeb48 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportService.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportService.java @@ -626,8 +626,13 @@ private void sendRequestInternal(final Transport.C } try { if (lifecycle.stoppedOrClosed()) { - // if we are not started the exception handling will remove the RequestHolder again and calls the handler to notify - // the caller. It will only notify if the toStop code hasn't done the work yet. + /* + * If we are not started the exception handling will remove the request holder again and calls the handler to notify the + * caller. It will only notify if toStop hasn't done the work yet. + * + * Do not edit this exception message, it is currently relied upon in production code! + */ + // TODO: make a dedicated exception for a stopped transport service? throw new TransportException("TransportService is closed stopped can't send request"); } if (timeoutHandler != null) { From ed7c4edc1d12f05330c93f968daaf11bdd0beb68 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Fri, 5 Apr 2019 09:55:57 -0400 Subject: [PATCH 2/7] Revert change used for testing --- server/src/main/java/org/elasticsearch/index/IndexService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/src/main/java/org/elasticsearch/index/IndexService.java b/server/src/main/java/org/elasticsearch/index/IndexService.java index 68966f6d0c946..1a49fd418735d 100644 --- a/server/src/main/java/org/elasticsearch/index/IndexService.java +++ b/server/src/main/java/org/elasticsearch/index/IndexService.java @@ -964,7 +964,7 @@ public String toString() { public static final Setting RETENTION_LEASE_SYNC_INTERVAL_SETTING = Setting.timeSetting( "index.soft_deletes.retention_lease.sync_interval", - new TimeValue(25, TimeUnit.MILLISECONDS), + new TimeValue(30, TimeUnit.SECONDS), new TimeValue(0, TimeUnit.MILLISECONDS), Property.Dynamic, Property.IndexScope); From eb493c384fbe81891d271bf04bfa9050a22310f4 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Fri, 5 Apr 2019 09:58:36 -0400 Subject: [PATCH 3/7] Update tests --- .../seqno/RetentionLeaseBackgroundSyncActionTests.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java b/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java index 6ad7d5039ae8b..ce1cf641ebb1b 100644 --- a/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java +++ b/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java @@ -42,6 +42,7 @@ import org.elasticsearch.test.transport.CapturingTransport; import org.elasticsearch.threadpool.TestThreadPool; import org.elasticsearch.threadpool.ThreadPool; +import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportService; import org.mockito.ArgumentCaptor; @@ -204,9 +205,15 @@ protected void doExecute(Task task, Request request, ActionListener captor = ArgumentCaptor.forClass(ParameterizedMessage.class); verify(retentionLeaseSyncActionLogger).warn(captor.capture(), same(e)); final ParameterizedMessage message = captor.getValue(); From 7539466a79af188f4ba91d4fa8964ea6c99cb7ce Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Fri, 5 Apr 2019 10:02:26 -0400 Subject: [PATCH 4/7] Fix test logic --- .../seqno/RetentionLeaseBackgroundSyncActionTests.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java b/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java index ce1cf641ebb1b..81ea56c609624 100644 --- a/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java +++ b/server/src/test/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncActionTests.java @@ -208,12 +208,10 @@ protected void doExecute(Task task, Request request, ActionListener captor = ArgumentCaptor.forClass(ParameterizedMessage.class); verify(retentionLeaseSyncActionLogger).warn(captor.capture(), same(e)); final ParameterizedMessage message = captor.getValue(); From 9ce0231b22e9b3a2c281e78e83ce5f658256ea73 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Fri, 5 Apr 2019 10:30:23 -0400 Subject: [PATCH 5/7] Refactor --- .../main/java/org/elasticsearch/ExceptionsHelper.java | 9 +++++++++ .../action/support/replication/ReplicationOperation.java | 7 +++---- .../index/seqno/RetentionLeaseBackgroundSyncAction.java | 6 +----- .../org/elasticsearch/transport/TransportService.java | 3 ++- 4 files changed, 15 insertions(+), 10 deletions(-) diff --git a/server/src/main/java/org/elasticsearch/ExceptionsHelper.java b/server/src/main/java/org/elasticsearch/ExceptionsHelper.java index e0525127ee7e7..e4269a375dd6c 100644 --- a/server/src/main/java/org/elasticsearch/ExceptionsHelper.java +++ b/server/src/main/java/org/elasticsearch/ExceptionsHelper.java @@ -29,6 +29,7 @@ import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException; import org.elasticsearch.index.Index; import org.elasticsearch.rest.RestStatus; +import org.elasticsearch.transport.TransportException; import java.io.IOException; import java.io.PrintWriter; @@ -193,6 +194,14 @@ public static Throwable unwrap(Throwable t, Class... clazzes) { return null; } + public static boolean isTransportStoppedForAction(final Throwable t, final String action) { + final TransportException maybeTransport = + (TransportException) ExceptionsHelper.unwrap(t, TransportException.class); + return maybeTransport != null + && (maybeTransport.getMessage().equals("TransportService is closed stopped can't send request") + || maybeTransport.getMessage().equals("transport stopped, action: " + action)); + } + /** * Throws the specified exception. If null if specified then true is returned. */ diff --git a/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java b/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java index 7fdb613c38bbf..5c338b4d3081e 100644 --- a/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java +++ b/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java @@ -205,10 +205,9 @@ public String toString() { private void onNoLongerPrimary(Exception failure) { final Throwable cause = ExceptionsHelper.unwrapCause(failure); - final boolean nodeIsClosing = cause instanceof NodeClosedException - || (cause instanceof TransportException && - ("TransportService is closed stopped can't send request".equals(cause.getMessage()) - || "transport stopped, action: internal:cluster/shard/failure".equals(cause.getMessage()))); + final boolean nodeIsClosing = + cause instanceof NodeClosedException + || ExceptionsHelper.isTransportStoppedForAction(cause, "internal:cluster/shard/failure"); final String message; if (nodeIsClosing) { message = String.format(Locale.ROOT, diff --git a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java index a28ec54d0441c..232e4ea196650 100644 --- a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java +++ b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java @@ -114,11 +114,7 @@ public void backgroundSync( ActionListener.wrap( r -> {}, e -> { - final TransportException maybeTransport = - (TransportException) ExceptionsHelper.unwrap(e, TransportException.class); - if (maybeTransport != null - && (maybeTransport.getMessage().equals("TransportService is closed stopped can't send request") - || maybeTransport.getMessage().equals("transport stopped, action: " + ACTION_NAME + "[p]"))) { + if (ExceptionsHelper.isTransportStoppedForAction(e, ACTION_NAME + "[p]")) { // we are likely shutting down return; } diff --git a/server/src/main/java/org/elasticsearch/transport/TransportService.java b/server/src/main/java/org/elasticsearch/transport/TransportService.java index 54cb7878aeb48..c8493edc97983 100644 --- a/server/src/main/java/org/elasticsearch/transport/TransportService.java +++ b/server/src/main/java/org/elasticsearch/transport/TransportService.java @@ -274,6 +274,7 @@ public void onFailure(Exception e) { } @Override public void doRun() { + // cf. ExceptionsHelper#isTransportStoppedForAction TransportException ex = new TransportException("transport stopped, action: " + holderToNotify.action()); holderToNotify.handler().handleException(ex); } @@ -632,7 +633,7 @@ private void sendRequestInternal(final Transport.C * * Do not edit this exception message, it is currently relied upon in production code! */ - // TODO: make a dedicated exception for a stopped transport service? + // TODO: make a dedicated exception for a stopped transport service? cf. ExceptionsHelper#isTransportStoppedForAction throw new TransportException("TransportService is closed stopped can't send request"); } if (timeoutHandler != null) { From 3d4eb5ee9a0d924f88c48caa6b56117154f1901e Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Fri, 5 Apr 2019 12:06:19 -0400 Subject: [PATCH 6/7] Fix checkstyle --- .../index/seqno/RetentionLeaseBackgroundSyncAction.java | 1 - 1 file changed, 1 deletion(-) diff --git a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java index 232e4ea196650..122db5799e9ac 100644 --- a/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java +++ b/server/src/main/java/org/elasticsearch/index/seqno/RetentionLeaseBackgroundSyncAction.java @@ -44,7 +44,6 @@ import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.indices.IndicesService; import org.elasticsearch.threadpool.ThreadPool; -import org.elasticsearch.transport.TransportException; import org.elasticsearch.transport.TransportService; import java.io.IOException; From 8b587b97f5b7c30195abe2674b5463d794cc77e4 Mon Sep 17 00:00:00 2001 From: Jason Tedor Date: Sat, 6 Apr 2019 00:27:16 -0400 Subject: [PATCH 7/7] checkstyle --- .../action/support/replication/ReplicationOperation.java | 1 - 1 file changed, 1 deletion(-) diff --git a/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java b/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java index 5c338b4d3081e..22e90cfc1356b 100644 --- a/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java +++ b/server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java @@ -38,7 +38,6 @@ import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.node.NodeClosedException; import org.elasticsearch.rest.RestStatus; -import org.elasticsearch.transport.TransportException; import java.io.IOException; import java.util.ArrayList;