assert transport exception

Author: dnhatn

server/src/main/java/org/elasticsearch/action/support/replication/ReplicationOperation.java

@@ -21,6 +21,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.store.AlreadyClosedException;
+import org.elasticsearch.Assertions;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionListener;
@@ -203,15 +204,20 @@ public String toString() {
     }
 
     private void onNoLongerPrimary(Exception failure) {
+        final boolean nodeIsClosing = failure instanceof NodeClosedException ||
+            (failure instanceof TransportException && "TransportService is closed stopped can't send request".equals(failure.getMessage()));
         final String message;
-        if (failure instanceof ShardStateAction.NoLongerPrimaryShardException) {
+        if (nodeIsClosing) {
+            message = String.format(Locale.ROOT, "primary node [%s] is shutting down while failing replica shard", primary.routingEntry());
+        } else {
+            if (Assertions.ENABLED) {
+                if (failure instanceof ShardStateAction.NoLongerPrimaryShardException == false) {
+                    throw new AssertionError("unexpected failure", failure);
+                }
+            }
             // we are no longer the primary, fail ourselves and start over
             message = String.format(Locale.ROOT, "primary shard [%s] was demoted while failing replica shard", primary.routingEntry());
             primary.failShard(message, failure);
-        } else {
-            // these can occur if the node is shutting down and are okay any other exception here is not expected and merits investigation.
-            assert failure instanceof NodeClosedException || failure instanceof TransportException : failure;
-            message = String.format(Locale.ROOT, "primary node [%s] is shutting down while failing replica shard", primary.routingEntry());
         }
         finishAsFailed(new RetryOnPrimaryException(primary.routingEntry().shardId(), message, failure));
     }

server/src/test/java/org/elasticsearch/action/support/replication/ReplicationOperationTests.java

@@ -43,6 +43,7 @@
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.node.NodeClosedException;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.TransportException;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -201,6 +202,8 @@ public void testNoLongerPrimary() throws Exception {
         final Exception shardActionFailure;
         if (randomBoolean()) {
             shardActionFailure = new NodeClosedException(new DiscoveryNode("foo", buildNewFakeTransportAddress(), Version.CURRENT));
+        } else if (randomBoolean()) {
+            shardActionFailure = new TransportException("TransportService is closed stopped can't send request");
         } else {
             shardActionFailure = new ShardStateAction.NoLongerPrimaryShardException(failedReplica.shardId(), "the king is dead");
         }

server/src/test/java/org/elasticsearch/action/support/replication/TransportWriteActionTests.java

@@ -335,11 +335,11 @@ public void testReplicaProxy() throws InterruptedException, ExecutionException {
             assertFalse(success.get());
             assertNotNull(failure.get());
         } else {
-            // simulated an "ignored" exception
+            // simulated a node closing exception
             transport.handleRemoteError(shardFailedRequest.requestId,
                 new NodeClosedException(state.nodes().getLocalNode()));
             assertFalse(success.get());
-            assertNull(failure.get());
+            assertNotNull(failure.get());
         }
     }