Cleanup TransportReplicationAction

core/src/main/java/org/elasticsearch/action/NoShardAvailableActionException.java

@@ -35,12 +35,12 @@ public NoShardAvailableActionException(ShardId shardId) {
         this(shardId, null);
     }
 
-    public NoShardAvailableActionException(ShardId shardId, String msg) {
-        this(shardId, msg, null);
+    public NoShardAvailableActionException(ShardId shardId, String msg, Object... args) {
+        this(shardId, msg, null, args);
     }
 
-    public NoShardAvailableActionException(ShardId shardId, String msg, Throwable cause) {
-        super(msg, cause);
+    public NoShardAvailableActionException(ShardId shardId, String msg, Throwable cause, Object... args) {
+        super(msg, cause, args);
         setShard(shardId);
     }
 

core/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

@@ -471,7 +471,7 @@ protected void shardOperationOnReplica(ShardId shardId, BulkShardRequest request
                     }
                     Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
                     if (update != null) {
-                        throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: " + update);
+                        throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: [{}]", update);
                     }
                     operation.execute(indexShard);
                     location = locationToSync(location, operation.getTranslogLocation());

core/src/main/java/org/elasticsearch/action/index/TransportIndexAction.java

@@ -195,7 +195,7 @@ protected void shardOperationOnReplica(ShardId shardId, IndexRequest request) {
         }
         Mapping update = operation.parsedDoc().dynamicMappingsUpdate();
         if (update != null) {
-            throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: " + update);
+            throw new RetryOnReplicaException(shardId, "Mappings are not available on the replica yet, triggered update: [{}]", update);
         }
         operation.execute(indexShard);
         processAfter(request, indexShard, operation.getTranslogLocation());

core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequest.java

@@ -24,10 +24,10 @@
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.WriteConsistencyLevel;
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.index.shard.ShardId;
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
@@ -41,12 +41,11 @@ public abstract class ReplicationRequest<T extends ReplicationRequest> extends A
 
     public static final TimeValue DEFAULT_TIMEOUT = new TimeValue(1, TimeUnit.MINUTES);
 
-    ShardId internalShardId;
+    ShardRouting internalShardRouting;
 
     protected TimeValue timeout = DEFAULT_TIMEOUT;
     protected String index;
 
-    private boolean threadedOperation = true;
     private WriteConsistencyLevel consistencyLevel = WriteConsistencyLevel.DEFAULT;
     private volatile boolean canHaveDuplicates = false;
 
@@ -76,7 +75,6 @@ protected ReplicationRequest(T request, ActionRequest originalRequest) {
         super(originalRequest);
         this.timeout = request.timeout();
         this.index = request.index();
-        this.threadedOperation = request.operationThreaded();
         this.consistencyLevel = request.consistencyLevel();
     }
 
@@ -91,23 +89,6 @@ public boolean canHaveDuplicates() {
         return canHaveDuplicates;
     }
 
-    /**
-     * Controls if the operation will be executed on a separate thread when executed locally.
-     */
-    public final boolean operationThreaded() {
-        return threadedOperation;
-    }
-
-    /**
-     * Controls if the operation will be executed on a separate thread when executed locally. Defaults
-     * to <tt>true</tt> when running in embedded mode.
-     */
-    @SuppressWarnings("unchecked")
-    public final T operationThreaded(boolean threadedOperation) {
-        this.threadedOperation = threadedOperation;
-        return (T) this;
-    }
-
     /**
      * A timeout to wait if the index operation can't be performed immediately. Defaults to <tt>1m</tt>.
      */
@@ -174,19 +155,18 @@ public ActionRequestValidationException validate() {
     public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
         if (in.readBoolean()) {
-            internalShardId = ShardId.readShardId(in);
+            internalShardRouting = ShardRouting.readShardRoutingEntry(in);
         }
         consistencyLevel = WriteConsistencyLevel.fromId(in.readByte());
         timeout = TimeValue.readTimeValue(in);
         index = in.readString();
         canHaveDuplicates = in.readBoolean();
-        // no need to serialize threaded* parameters, since they only matter locally
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
-        out.writeOptionalStreamable(internalShardId);
+        out.writeOptionalStreamable(internalShardRouting);
         out.writeByte(consistencyLevel.id());
         timeout.writeTo(out);
         out.writeString(index);

core/src/main/java/org/elasticsearch/action/support/replication/ReplicationRequestBuilder.java

@@ -35,16 +35,6 @@ protected ReplicationRequestBuilder(ElasticsearchClient client, Action<Request,
         super(client, action, request);
     }
 
-    /**
-     * Controls if the operation will be executed on a separate thread when executed locally. Defaults
-     * to <tt>true</tt> when running in embedded mode.
-     */
-    @SuppressWarnings("unchecked")
-    public final RequestBuilder setOperationThreaded(boolean threadedOperation) {
-        request.operationThreaded(threadedOperation);
-        return (RequestBuilder) this;
-    }
-
     /**
      * A timeout to wait if the index operation can't be performed immediately. Defaults to <tt>1m</tt>.
      */