Delete unowned documents during split

Author: lkts

server/src/main/java/org/elasticsearch/cluster/metadata/IndexReshardingMetadata.java

@@ -207,6 +207,10 @@ public static IndexReshardingMetadata newSplitByMultiple(int shardCount, int mul
         return new IndexReshardingMetadata(IndexReshardingState.Split.newSplitByMultiple(shardCount, multiple));
     }
 
+    public static boolean isSplitSource(ShardId shardId, @Nullable IndexReshardingMetadata reshardingMetadata) {
+        return reshardingMetadata != null && reshardingMetadata.isSplit() && reshardingMetadata.getSplit().isSourceShard(shardId.id());
+    }
+
     public static boolean isSplitTarget(ShardId shardId, @Nullable IndexReshardingMetadata reshardingMetadata) {
         return reshardingMetadata != null && reshardingMetadata.isSplit() && reshardingMetadata.getSplit().isTargetShard(shardId.id());
     }
@@ -221,6 +225,16 @@ public IndexReshardingMetadata transitionSplitTargetToNewState(
         return new IndexReshardingMetadata(builder.build());
     }
 
+    public IndexReshardingMetadata transitionSplitSourceToNewState(
+        ShardId shardId,
+        IndexReshardingState.Split.SourceShardState newSourceState
+    ) {
+        assert state instanceof IndexReshardingState.Split;
+        IndexReshardingState.Split.Builder builder = new IndexReshardingState.Split.Builder((IndexReshardingState.Split) state);
+        builder.setSourceShardState(shardId.getId(), newSourceState);
+        return new IndexReshardingMetadata(builder.build());
+    }
+
     /**
      * @return the split state of this metadata block, or throw IllegalArgumentException if this metadata doesn't represent a split
      */

server/src/main/java/org/elasticsearch/cluster/metadata/IndexReshardingState.java

@@ -21,8 +21,11 @@
 
 import java.io.IOException;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 /**
@@ -350,8 +353,12 @@ public SourceShardState getSourceShardState(int shardNum) {
             return sourceShards[shardNum];
         }
 
+        public boolean isSourceShard(int shardId) {
+            return shardId < shardCountBefore();
+        }
+
         public boolean isTargetShard(int shardId) {
-            return shardId >= shardCountBefore();
+            return isSourceShard(shardId) == false;
         }
 
         /**
@@ -389,6 +396,10 @@ public Stream<TargetShardState> targetStates() {
             return Arrays.stream(targetShards);
         }
 
+        public Stream<SourceShardState> sourceStates() {
+            return Arrays.stream(sourceShards);
+        }
+
         /**
          * Check whether all target shards for the given source shard are done.
          * @param shardNum a source shard index greater than or equal to 0 and less than the original shard count

server/src/main/java/org/elasticsearch/indices/cluster/IndicesClusterStateService.java

@@ -1004,7 +1004,7 @@ private static DiscoveryNode findSourceNodeForReshardSplitRecovery(
         ShardRouting sourceShardRouting = routingTable.shardRoutingTable(sourceShardId).primaryShard();
 
         if (sourceShardRouting.active() == false) {
-            assert false : sourceShardRouting;
+            assert false : sourceShardRouting.shortSummary();
             logger.trace("can't find reshard split source node because source shard {} is not active.", sourceShardRouting);
             return null;
         }
@@ -1014,7 +1014,7 @@ private static DiscoveryNode findSourceNodeForReshardSplitRecovery(
             assert false : "Source node for reshard does not exist: " + sourceShardRouting.currentNodeId();
             logger.trace(
                 "can't find reshard split source node because source shard {} is assigned to an unknown node.",
-                sourceShardRouting
+                sourceShardRouting.shortSummary()
             );
             return null;
         }