Revert "[RCI] Check blocks while having index shard permit in TransportReplicationAction (elastic#35332)"

This reverts commit 31567ce.

Author: tlrx

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

@@ -235,39 +235,9 @@ protected TransportRequestOptions transportOptions(Settings settings) {
  return TransportRequestOptions.EMPTY;
  }
 
- private String concreteIndex(final ClusterState state, final ReplicationRequest request) {
- return resolveIndex() ? indexNameExpressionResolver.concreteSingleIndex(state, request).getName() : request.index();
- }
-
- private ClusterBlockException blockExceptions(final ClusterState state, final String indexName) {
- ClusterBlockLevel globalBlockLevel = globalBlockLevel();
- if (globalBlockLevel != null) {
- ClusterBlockException blockException = state.blocks().globalBlockedException(globalBlockLevel);
- if (blockException != null) {
- return blockException;
- }
- }
- ClusterBlockLevel indexBlockLevel = indexBlockLevel();
- if (indexBlockLevel != null) {
- ClusterBlockException blockException = state.blocks().indexBlockedException(indexBlockLevel, indexName);
- if (blockException != null) {
- return blockException;
- }
- }
- return null;
- }
-
  protected boolean retryPrimaryException(final Throwable e) {
  return e.getClass() == ReplicationOperation.RetryOnPrimaryException.class
- || TransportActions.isShardNotAvailableException(e)
- || isRetryableClusterBlockException(e);
- }
-
- boolean isRetryableClusterBlockException(final Throwable e) {
- if (e instanceof ClusterBlockException) {
- return ((ClusterBlockException) e).retryable();
- }
- return false;
+ || TransportActions.isShardNotAvailableException(e);
  }
 
  protected class OperationTransportHandler implements TransportRequestHandler<Request> {
@@ -340,15 +310,6 @@ protected void doRun() throws Exception {
  @Override
  public void onResponse(PrimaryShardReference primaryShardReference) {
  try {
- final ClusterState clusterState = clusterService.state();
- final IndexMetaData indexMetaData = clusterState.metaData().getIndexSafe(primaryShardReference.routingEntry().index());
-
- final ClusterBlockException blockException = blockExceptions(clusterState, indexMetaData.getIndex().getName());
- if (blockException != null) {
- logger.trace("cluster is blocked, action failed on primary", blockException);
- throw blockException;
- }
-
  if (primaryShardReference.isRelocated()) {
  primaryShardReference.close(); // release shard operation lock as soon as possible
  setPhase(replicationTask, "primary_delegation");
@@ -362,7 +323,7 @@ public void onResponse(PrimaryShardReference primaryShardReference) {
  response.readFrom(in);
  return response;
  };
- DiscoveryNode relocatingNode = clusterState.nodes().get(primary.relocatingNodeId());
+ DiscoveryNode relocatingNode = clusterService.state().nodes().get(primary.relocatingNodeId());
  transportService.sendRequest(relocatingNode, transportPrimaryAction,
  new ConcreteShardRequest<>(request, primary.allocationId().getRelocationId(), primaryTerm),
  transportOptions,
@@ -735,42 +696,35 @@ public void onFailure(Exception e) {
  protected void doRun() {
  setPhase(task, "routing");
  final ClusterState state = observer.setAndGetObservedState();
- final String concreteIndex = concreteIndex(state, request);
- final ClusterBlockException blockException = blockExceptions(state, concreteIndex);
- if (blockException != null) {
- if (blockException.retryable()) {
- logger.trace("cluster is blocked, scheduling a retry", blockException);
- retry(blockException);
- } else {
- finishAsFailed(blockException);
- }
- } else {
- // request does not have a shardId yet, we need to pass the concrete index to resolve shardId
- final IndexMetaData indexMetaData = state.metaData().index(concreteIndex);
- if (indexMetaData == null) {
- retry(new IndexNotFoundException(concreteIndex));
- return;
- }
- if (indexMetaData.getState() == IndexMetaData.State.CLOSE) {
- throw new IndexClosedException(indexMetaData.getIndex());
- }
+ if (handleBlockExceptions(state)) {
+ return;
+ }
+
+ // request does not have a shardId yet, we need to pass the concrete index to resolve shardId
+ final String concreteIndex = concreteIndex(state);
+ final IndexMetaData indexMetaData = state.metaData().index(concreteIndex);
+ if (indexMetaData == null) {
+ retry(new IndexNotFoundException(concreteIndex));
+ return;
+ }
+ if (indexMetaData.getState() == IndexMetaData.State.CLOSE) {
+ throw new IndexClosedException(indexMetaData.getIndex());
+ }
 
- // resolve all derived request fields, so we can route and apply it
- resolveRequest(indexMetaData, request);
- assert request.shardId() != null : "request shardId must be set in resolveRequest";
- assert request.waitForActiveShards() != ActiveShardCount.DEFAULT :
- "request waitForActiveShards must be set in resolveRequest";
+ // resolve all derived request fields, so we can route and apply it
+ resolveRequest(indexMetaData, request);
+ assert request.shardId() != null : "request shardId must be set in resolveRequest";
+ assert request.waitForActiveShards() != ActiveShardCount.DEFAULT : "request waitForActiveShards must be set in resolveRequest";
 
- final ShardRouting primary = primary(state);
- if (retryIfUnavailable(state, primary)) {
- return;
- }
- final DiscoveryNode node = state.nodes().get(primary.currentNodeId());
- if (primary.currentNodeId().equals(state.nodes().getLocalNodeId())) {
- performLocalAction(state, primary, node, indexMetaData);
- } else {
- performRemoteAction(state, primary, node);
- }
+ final ShardRouting primary = primary(state);
+ if (retryIfUnavailable(state, primary)) {
+ return;
+ }
+ final DiscoveryNode node = state.nodes().get(primary.currentNodeId());
+ if (primary.currentNodeId().equals(state.nodes().getLocalNodeId())) {
+ performLocalAction(state, primary, node, indexMetaData);
+ } else {
+ performRemoteAction(state, primary, node);
  }
  }
 
@@ -822,11 +776,44 @@ private boolean retryIfUnavailable(ClusterState state, ShardRouting primary) {
  return false;
  }
 
+ private String concreteIndex(ClusterState state) {
+ return resolveIndex() ? indexNameExpressionResolver.concreteSingleIndex(state, request).getName() : request.index();
+ }
+
  private ShardRouting primary(ClusterState state) {
  IndexShardRoutingTable indexShard = state.getRoutingTable().shardRoutingTable(request.shardId());
  return indexShard.primaryShard();
  }
 
+ private boolean handleBlockExceptions(ClusterState state) {
+ ClusterBlockLevel globalBlockLevel = globalBlockLevel();
+ if (globalBlockLevel != null) {
+ ClusterBlockException blockException = state.blocks().globalBlockedException(globalBlockLevel);
+ if (blockException != null) {
+ handleBlockException(blockException);
+ return true;
+ }
+ }
+ ClusterBlockLevel indexBlockLevel = indexBlockLevel();
+ if (indexBlockLevel != null) {
+ ClusterBlockException blockException = state.blocks().indexBlockedException(indexBlockLevel, concreteIndex(state));
+ if (blockException != null) {
+ handleBlockException(blockException);
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private void handleBlockException(ClusterBlockException blockException) {
+ if (blockException.retryable()) {
+ logger.trace("cluster is blocked, scheduling a retry", blockException);
+ retry(blockException);
+ } else {
+ finishAsFailed(blockException);
+ }
+ }
+
  private void performAction(final DiscoveryNode node, final String action, final boolean isPrimaryAction,
  final TransportRequest requestToPerform) {
  transportService.sendRequest(node, action, requestToPerform, transportOptions, new TransportResponseHandler<Response>() {