diff --git a/buildSrc/src/main/resources/checkstyle_suppressions.xml b/buildSrc/src/main/resources/checkstyle_suppressions.xml
index c77ef0f9d5b73..10387db0530e3 100644
--- a/buildSrc/src/main/resources/checkstyle_suppressions.xml
+++ b/buildSrc/src/main/resources/checkstyle_suppressions.xml
@@ -19,7 +19,6 @@
-
@@ -101,7 +100,6 @@
-
diff --git a/core/src/main/java/org/elasticsearch/action/DocWriteResponse.java b/core/src/main/java/org/elasticsearch/action/DocWriteResponse.java
index 4df43b75401f9..0925c7441446a 100644
--- a/core/src/main/java/org/elasticsearch/action/DocWriteResponse.java
+++ b/core/src/main/java/org/elasticsearch/action/DocWriteResponse.java
@@ -18,10 +18,15 @@
*/
package org.elasticsearch.action;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.action.support.WriteResponse;
+import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.xcontent.StatusToXContent;
import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.index.IndexSettings;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.rest.RestStatus;
@@ -30,12 +35,13 @@
/**
* A base class for the response of a write operation that involves a single doc
*/
-public abstract class DocWriteResponse extends ReplicationResponse implements StatusToXContent {
+public abstract class DocWriteResponse extends ReplicationResponse implements WriteResponse, StatusToXContent {
private ShardId shardId;
private String id;
private String type;
private long version;
+ private boolean forcedRefresh;
public DocWriteResponse(ShardId shardId, String type, String id, long version) {
this.shardId = shardId;
@@ -84,6 +90,20 @@ public long getVersion() {
return this.version;
}
+ /**
+ * Did this request force a refresh? Requests that set {@link WriteRequest#setRefreshPolicy(RefreshPolicy)} to
+ * {@link RefreshPolicy#IMMEDIATE} will always return true for this. Requests that set it to {@link RefreshPolicy#WAIT_UNTIL} will
+ * only return true here if they run out of refresh listener slots (see {@link IndexSettings#MAX_REFRESH_LISTENERS_PER_SHARD}).
+ */
+ public boolean forcedRefresh() {
+ return forcedRefresh;
+ }
+
+ @Override
+ public void setForcedRefresh(boolean forcedRefresh) {
+ this.forcedRefresh = forcedRefresh;
+ }
+
/** returns the rest status for this response (based on {@link ShardInfo#status()} */
public RestStatus status() {
return getShardInfo().status();
@@ -97,6 +117,7 @@ public void readFrom(StreamInput in) throws IOException {
type = in.readString();
id = in.readString();
version = in.readZLong();
+ forcedRefresh = in.readBoolean();
}
@Override
@@ -106,6 +127,7 @@ public void writeTo(StreamOutput out) throws IOException {
out.writeString(type);
out.writeString(id);
out.writeZLong(version);
+ out.writeBoolean(forcedRefresh);
}
static final class Fields {
@@ -121,7 +143,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
builder.field(Fields._INDEX, shardId.getIndexName())
.field(Fields._TYPE, type)
.field(Fields._ID, id)
- .field(Fields._VERSION, version);
+ .field(Fields._VERSION, version)
+ .field("forced_refresh", forcedRefresh);
shardInfo.toXContent(builder, params);
return builder;
}
diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java
index 8bb124d8fc494..a29918b438ef3 100644
--- a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java
+++ b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportFlushAction.java
@@ -19,9 +19,9 @@
package org.elasticsearch.action.admin.indices.flush;
-import org.elasticsearch.action.ReplicationResponse;
import org.elasticsearch.action.ShardOperationFailedException;
import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.support.replication.TransportBroadcastReplicationAction;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.service.ClusterService;
diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java
index 7e750b9767790..82fb6d70ca441 100644
--- a/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java
+++ b/core/src/main/java/org/elasticsearch/action/admin/indices/flush/TransportShardFlushAction.java
@@ -19,14 +19,13 @@
package org.elasticsearch.action.admin.indices.flush;
-import org.elasticsearch.action.ReplicationResponse;
import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.support.replication.TransportReplicationAction;
import org.elasticsearch.cluster.action.shard.ShardStateAction;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.shard.IndexShard;
@@ -55,18 +54,19 @@ protected ReplicationResponse newResponseInstance() {
}
@Override
- protected Tuple shardOperationOnPrimary(ShardFlushRequest shardRequest) {
+ protected PrimaryResult shardOperationOnPrimary(ShardFlushRequest shardRequest) {
IndexShard indexShard = indicesService.indexServiceSafe(shardRequest.shardId().getIndex()).getShard(shardRequest.shardId().id());
indexShard.flush(shardRequest.getRequest());
logger.trace("{} flush request executed on primary", indexShard.shardId());
- return new Tuple<>(new ReplicationResponse(), shardRequest);
+ return new PrimaryResult(shardRequest, new ReplicationResponse());
}
@Override
- protected void shardOperationOnReplica(ShardFlushRequest request) {
+ protected ReplicaResult shardOperationOnReplica(ShardFlushRequest request) {
IndexShard indexShard = indicesService.indexServiceSafe(request.shardId().getIndex()).getShard(request.shardId().id());
indexShard.flush(request.getRequest());
logger.trace("{} flush request executed on replica", indexShard.shardId());
+ return new ReplicaResult();
}
@Override
diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java
index 34bf39daabd19..ac64e276778f6 100644
--- a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java
+++ b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportRefreshAction.java
@@ -19,10 +19,10 @@
package org.elasticsearch.action.admin.indices.refresh;
-import org.elasticsearch.action.ReplicationResponse;
import org.elasticsearch.action.ShardOperationFailedException;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.replication.BasicReplicationRequest;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.support.replication.TransportBroadcastReplicationAction;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.service.ClusterService;
diff --git a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java
index 0670c1f3cc64d..d7d0c289953a4 100644
--- a/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java
+++ b/core/src/main/java/org/elasticsearch/action/admin/indices/refresh/TransportShardRefreshAction.java
@@ -19,15 +19,14 @@
package org.elasticsearch.action.admin.indices.refresh;
-import org.elasticsearch.action.ReplicationResponse;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.replication.BasicReplicationRequest;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.support.replication.TransportReplicationAction;
import org.elasticsearch.cluster.action.shard.ShardStateAction;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.service.ClusterService;
-import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.shard.IndexShard;
@@ -36,10 +35,8 @@
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportService;
-/**
- *
- */
-public class TransportShardRefreshAction extends TransportReplicationAction {
+public class TransportShardRefreshAction
+ extends TransportReplicationAction {
public static final String NAME = RefreshAction.NAME + "[s]";
@@ -47,8 +44,8 @@ public class TransportShardRefreshAction extends TransportReplicationAction shardOperationOnPrimary(BasicReplicationRequest shardRequest) {
+ protected PrimaryResult shardOperationOnPrimary(BasicReplicationRequest shardRequest) {
IndexShard indexShard = indicesService.indexServiceSafe(shardRequest.shardId().getIndex()).getShard(shardRequest.shardId().id());
indexShard.refresh("api");
logger.trace("{} refresh request executed on primary", indexShard.shardId());
- return new Tuple<>(new ReplicationResponse(), shardRequest);
+ return new PrimaryResult(shardRequest, new ReplicationResponse());
}
@Override
- protected void shardOperationOnReplica(BasicReplicationRequest request) {
+ protected ReplicaResult shardOperationOnReplica(BasicReplicationRequest request) {
final ShardId shardId = request.shardId();
IndexShard indexShard = indicesService.indexServiceSafe(shardId.getIndex()).getShard(shardId.id());
indexShard.refresh("api");
logger.trace("{} refresh request executed on replica", indexShard.shardId());
+ return new ReplicaResult();
}
@Override
diff --git a/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java b/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
index a5775656475e4..85d7147ada084 100644
--- a/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
+++ b/core/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java
@@ -26,6 +26,7 @@
import org.elasticsearch.action.WriteConsistencyLevel;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Strings;
@@ -54,16 +55,21 @@
* Note that we only support refresh on the bulk request not per item.
* @see org.elasticsearch.client.Client#bulk(BulkRequest)
*/
-public class BulkRequest extends ActionRequest implements CompositeIndicesRequest {
+public class BulkRequest extends ActionRequest implements CompositeIndicesRequest, WriteRequest {
private static final int REQUEST_OVERHEAD = 50;
+ /**
+ * Requests that are part of this request. It is only possible to add things that are both {@link ActionRequest}s and
+ * {@link WriteRequest}s to this but java doesn't support syntax to declare that everything in the array has both types so we declare
+ * the one with the least casts.
+ */
final List> requests = new ArrayList<>();
List