import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
public void updateCheckpointForShard(ShardId shardId) {
    execute(new PrimaryRequest(shardId), new ActionListener<ReplicationResponse>() {
        @Override
        public void onResponse(ReplicationResponse replicationResponse) {
            if (logger.isTraceEnabled()) {
                logger.trace("{} global checkpoint successfully updated (shard info [{}])", shardId,
                    replicationResponse.getShardInfo());
            }
        }

        @Override
        public void onFailure(Exception e) {
            logger.debug((Supplier<?>) () -> new ParameterizedMessage("{} failed to update global checkpoint", shardId), e);
        }
    });
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
private static Set<ShardId> findAllShardsForIndex(Path indexPath, Index index) throws IOException {
    assert indexPath.getFileName().toString().equals(index.getUUID());
    Set<ShardId> shardIds = new HashSet<>();
    if (Files.isDirectory(indexPath)) {
        try (DirectoryStream<Path> stream = Files.newDirectoryStream(indexPath)) {
            for (Path shardPath : stream) {
                String fileName = shardPath.getFileName().toString();
                if (Files.isDirectory(shardPath) && fileName.chars().allMatch(Character::isDigit)) {
                    int shardId = Integer.parseInt(fileName);
                    ShardId id = new ShardId(index, shardId);
                    shardIds.add(id);
                }
            }
        }
    }
    return shardIds;
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
/**
 * Tries to lock all local shards for the given index. If any of the shard locks can't be acquired
 * an {@link LockObtainFailedException} is thrown and all previously acquired locks are released.
 *
 * @param index the index to lock shards for
 * @param lockTimeoutMS how long to wait for acquiring the indices shard locks
 * @return the {@link ShardLock} instances for this index.
 * @throws IOException if an IOException occurs.
 */
public List<ShardLock> lockAllForIndex(Index index, Settings settings, long lockTimeoutMS) throws IOException {
    final Integer numShards = settings.getAsInt(IndexMetaData.SETTING_NUMBER_OF_SHARDS, null);
    if (numShards == null || numShards <= 0) {
        throw new IllegalArgumentException("settings must contain a non-null > 0 number of shards");
    }
    logger.trace("locking all shards for index {} - [{}]", index, numShards);
    List<ShardLock> allLocks = new ArrayList<>(numShards);
    boolean success = false;
    long startTimeNS = System.nanoTime();
    try {
        for (int i = 0; i < numShards; i++) {
            long timeoutLeftMS = Math.max(0, lockTimeoutMS - TimeValue.nsecToMSec((System.nanoTime() - startTimeNS)));
            allLocks.add(shardLock(new ShardId(index, i), timeoutLeftMS));
        }
        success = true;
    } finally {
        if (success == false) {
            logger.trace("unable to lock all shards for index {}", index);
            IOUtils.closeWhileHandlingException(allLocks);
        }
    }
    return allLocks;
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
private static Set<ShardId> findAllShardsForIndex(Path indexPath) throws IOException {
    Set<ShardId> shardIds = new HashSet<>();
    if (Files.isDirectory(indexPath)) {
        try (DirectoryStream<Path> stream = Files.newDirectoryStream(indexPath)) {
            String currentIndex = indexPath.getFileName().toString();
            for (Path shardPath : stream) {
                if (Files.isDirectory(shardPath)) {
                    Integer shardId = Ints.tryParse(shardPath.getFileName().toString());
                    if (shardId != null) {
                        ShardId id = new ShardId(currentIndex, shardId);
                        shardIds.add(id);
                    }
                }
            }
        }
    }
    return shardIds;
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
@Override
protected FieldStatsShardResponse shardOperation(FieldStatsShardRequest request) {
    ShardId shardId = request.shardId();
    Map<String, FieldStats> fieldStats = new HashMap<>();
    IndexService indexServices = indicesService.indexServiceSafe(shardId.getIndex());
    MapperService mapperService = indexServices.mapperService();
    IndexShard shard = indexServices.shardSafe(shardId.id());
    try (Engine.Searcher searcher = shard.acquireSearcher("fieldstats")) {
        for (String field : request.getFields()) {
            MappedFieldType fieldType = mapperService.fullName(field);
            if (fieldType != null) {
                IndexReader reader = searcher.reader();
                Terms terms = MultiFields.getTerms(reader, field);
                if (terms != null) {
                    fieldStats.put(field, fieldType.stats(terms, reader.maxDoc()));
                }
            } else {
                throw new IllegalArgumentException("field [" + field + "] doesn't exist");
            }
        }
    } catch (IOException e) {
        throw ExceptionsHelper.convertToElastic(e);
    }
    return new FieldStatsShardResponse(shardId, fieldStats);
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
public void testAddedReplicaAfterPrimaryOperation() throws Exception {
    final String index = "test";
    final ShardId shardId = new ShardId(index, "_na_", 0);
    final ClusterState initialState = stateWithActivePrimary(index, true, 0);
    final ClusterState stateWithAddedReplicas;
    if (randomBoolean()) {
        stateWithAddedReplicas = state(index, true, ShardRoutingState.STARTED,
            randomBoolean() ? ShardRoutingState.INITIALIZING : ShardRoutingState.STARTED);
    } else {
        stateWithAddedReplicas = state(index, true, ShardRoutingState.RELOCATING);
    }
    testClusterStateChangeAfterPrimaryOperation(shardId, initialState, stateWithAddedReplicas);
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
@Inject
public BlobRecoverySource(Settings settings, TransportService transportService, IndicesService indicesService,
                          RecoverySettings recoverySettings, ClusterService clusterService,
                          BlobTransferTarget blobTransferTarget, BlobIndices blobIndices) {
    super(settings);
    this.transportService = transportService;
    this.indicesService = indicesService;
    this.clusterService = clusterService;
    this.blobTransferTarget = blobTransferTarget;
    this.blobIndices = blobIndices;
    this.indicesService.indicesLifecycle().addListener(new IndicesLifecycle.Listener() {
        @Override
        public void beforeIndexShardClosed(ShardId shardId, @Nullable IndexShard indexShard,
                                           Settings indexSettings) {
            if (indexShard != null) {
                ongoingRecoveries.cancel(indexShard, "shard is closed");
            }
        }
    });

    this.recoverySettings = recoverySettings;

}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
private void writeIndex(NodeEnvironment nodeEnv, IndexSettings indexSettings,
                        int numIdxFiles, int numTranslogFiles) throws IOException {
    NodeEnvironment.NodePath[] nodePaths = nodeEnv.nodePaths();
    Path[] oldIndexPaths = new Path[nodePaths.length];
    for (int i = 0; i < nodePaths.length; i++) {
        oldIndexPaths[i] = nodePaths[i].indicesPath.resolve(indexSettings.getIndex().getName());
    }
    IndexMetaData.FORMAT.write(indexSettings.getIndexMetaData(), oldIndexPaths);
    for (int id = 0; id < indexSettings.getNumberOfShards(); id++) {
        Path oldIndexPath = randomFrom(oldIndexPaths);
        ShardId shardId = new ShardId(indexSettings.getIndex(), id);
        if (indexSettings.hasCustomDataPath()) {
            Path customIndexPath = nodeEnv.resolveBaseCustomLocation(indexSettings).resolve(indexSettings.getIndex().getName());
            writeShard(shardId, customIndexPath, numIdxFiles, numTranslogFiles);
        } else {
            writeShard(shardId, oldIndexPath, numIdxFiles, numTranslogFiles);
        }
        ShardStateMetaData state = new ShardStateMetaData(true, indexSettings.getUUID(), AllocationId.newInitializing());
        ShardStateMetaData.FORMAT.write(state, oldIndexPath.resolve(String.valueOf(shardId.getId())));
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
@Override
public void postIndex(ShardId shardId, Engine.Index indexOperation, Engine.IndexResult result) {
    if (result.hasFailure() == false) {
        final ParsedDocument doc = indexOperation.parsedDoc();
        final long tookInNanos = result.getTook();
        if (indexWarnThreshold >= 0 && tookInNanos > indexWarnThreshold) {
            indexLogger.warn("{}", new SlowLogParsedDocumentPrinter(index, doc, tookInNanos, reformat, maxSourceCharsToLog));
        } else if (indexInfoThreshold >= 0 && tookInNanos > indexInfoThreshold) {
            indexLogger.info("{}", new SlowLogParsedDocumentPrinter(index, doc, tookInNanos, reformat, maxSourceCharsToLog));
        } else if (indexDebugThreshold >= 0 && tookInNanos > indexDebugThreshold) {
            indexLogger.debug("{}", new SlowLogParsedDocumentPrinter(index, doc, tookInNanos, reformat, maxSourceCharsToLog));
        } else if (indexTraceThreshold >= 0 && tookInNanos > indexTraceThreshold) {
            indexLogger.trace("{}", new SlowLogParsedDocumentPrinter(index, doc, tookInNanos, reformat, maxSourceCharsToLog));
        }
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
public void testGenerateShardId() {
    int[][] possibleValues = new int[][] {
        {8,4,2}, {20, 10, 2}, {36, 12, 3}, {15,5,1}
    };
    for (int i = 0; i < 10; i++) {
        int[] shardSplits = randomFrom(possibleValues);
        assertEquals(shardSplits[0], (shardSplits[0] / shardSplits[1]) * shardSplits[1]);
        assertEquals(shardSplits[1], (shardSplits[1] / shardSplits[2]) * shardSplits[2]);
        IndexMetaData metaData = IndexMetaData.builder("test").settings(settings(Version.CURRENT)).numberOfShards(shardSplits[0])
            .numberOfReplicas(1).build();
        String term = randomAsciiOfLength(10);
        final int shard = OperationRouting.generateShardId(metaData, term, null);
        IndexMetaData shrunk = IndexMetaData.builder("test").settings(settings(Version.CURRENT)).numberOfShards(shardSplits[1])
            .numberOfReplicas(1)
            .setRoutingNumShards(shardSplits[0]).build();
        int shrunkShard = OperationRouting.generateShardId(shrunk, term, null);
        Set<ShardId> shardIds = IndexMetaData.selectShrinkShards(shrunkShard, metaData, shrunk.getNumberOfShards());
        assertEquals(1, shardIds.stream().filter((sid) -> sid.id() == shard).count());

        shrunk = IndexMetaData.builder("test").settings(settings(Version.CURRENT)).numberOfShards(shardSplits[2]).numberOfReplicas(1)
            .setRoutingNumShards(shardSplits[0]).build();
        shrunkShard = OperationRouting.generateShardId(shrunk, term, null);
        shardIds = IndexMetaData.selectShrinkShards(shrunkShard, metaData, shrunk.getNumberOfShards());
        assertEquals(Arrays.toString(shardSplits), 1, shardIds.stream().filter((sid) -> sid.id() == shard).count());
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
private void processShardRouting(Map<String, Map<String, List<Integer>>> routing, ShardRouting shardRouting, ShardId shardId) {
    String node;
    int id;
    String index = shardId.getIndex();

    if (shardRouting == null) {
        node = service.localNode().id();
        id = UnassignedShard.markUnassigned(shardId.id());
    } else {
        node = shardRouting.currentNodeId();
        id = shardRouting.id();
    }
    Map<String, List<Integer>> nodeMap = routing.get(node);
    if (nodeMap == null) {
        nodeMap = new TreeMap<>();
        routing.put(node, nodeMap);
    }

    List<Integer> shards = nodeMap.get(index);
    if (shards == null) {
        shards = new ArrayList<>();
        nodeMap.put(index, shards);
    }
    shards.add(id);
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
/**
 * Check if any of the indices to be closed are currently being restored from a snapshot and fail closing if such an index
 * is found as closing an index that is being restored makes the index unusable (it cannot be recovered).
 */
public static void checkIndexClosing(ClusterState currentState, Set<IndexMetaData> indices) {
    RestoreInProgress restore = currentState.custom(RestoreInProgress.TYPE);
    if (restore != null) {
        Set<Index> indicesToFail = null;
        for (RestoreInProgress.Entry entry : restore.entries()) {
            for (ObjectObjectCursor<ShardId, RestoreInProgress.ShardRestoreStatus> shard : entry.shards()) {
                if (!shard.value.state().completed()) {
                    IndexMetaData indexMetaData = currentState.metaData().index(shard.key.getIndex());
                    if (indexMetaData != null && indices.contains(indexMetaData)) {
                        if (indicesToFail == null) {
                            indicesToFail = new HashSet<>();
                        }
                        indicesToFail.add(shard.key.getIndex());
                    }
                }
            }
        }
        if (indicesToFail != null) {
            throw new IllegalArgumentException("Cannot close indices that are being restored: " + indicesToFail);
        }
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
private Entry randomSnapshot() {
    Snapshot snapshot = new Snapshot(randomAsciiOfLength(10), new SnapshotId(randomAsciiOfLength(10), randomAsciiOfLength(10)));
    boolean includeGlobalState = randomBoolean();
    boolean partial = randomBoolean();
    State state = randomFrom(State.values());
    int numberOfIndices = randomIntBetween(0, 10);
    List<IndexId> indices = new ArrayList<>();
    for (int i = 0; i < numberOfIndices; i++) {
        indices.add(new IndexId(randomAsciiOfLength(10), randomAsciiOfLength(10)));
    }
    long startTime = randomLong();
    long repositoryStateId = randomLong();
    ImmutableOpenMap.Builder<ShardId, SnapshotsInProgress.ShardSnapshotStatus> builder = ImmutableOpenMap.builder();
    int shardsCount = randomIntBetween(0, 10);
    for (int j = 0; j < shardsCount; j++) {
        ShardId shardId = new ShardId(new Index(randomAsciiOfLength(10), randomAsciiOfLength(10)), randomIntBetween(0, 10));
        String nodeId = randomAsciiOfLength(10);
        State shardState = randomFrom(State.values());
        builder.put(shardId, new SnapshotsInProgress.ShardSnapshotStatus(nodeId, shardState));
    }
    ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus> shards = builder.build();
    return new Entry(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards);
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
/**
 * cancel all ongoing recoveries for the given shard
 *
 * @param reason       reason for cancellation
 * @param shardId      shardId for which to cancel recoveries
 * @return true if a recovery was cancelled
 */
public boolean cancelRecoveriesForShard(ShardId shardId, String reason) {
    boolean cancelled = false;
    List<RecoveryTarget> matchedRecoveries = new ArrayList<>();
    synchronized (onGoingRecoveries) {
        for (Iterator<RecoveryTarget> it = onGoingRecoveries.values().iterator(); it.hasNext(); ) {
            RecoveryTarget status = it.next();
            if (status.shardId().equals(shardId)) {
                matchedRecoveries.add(status);
                it.remove();
            }
        }
    }
    for (RecoveryTarget removed : matchedRecoveries) {
        logger.trace("{} canceled recovery from {}, id [{}] (reason [{}])",
            removed.shardId(), removed.sourceNode(), removed.recoveryId(), reason);
        removed.cancel(reason);
        cancelled = true;
    }
    return cancelled;
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
public static RestoreInProgress.State overallState(RestoreInProgress.State nonCompletedState,
                                                   ImmutableOpenMap<ShardId, RestoreInProgress.ShardRestoreStatus> shards) {
    boolean hasFailed = false;
    for (ObjectCursor<RestoreInProgress.ShardRestoreStatus> status : shards.values()) {
        if (!status.value.state().completed()) {
            return nonCompletedState;
        }
        if (status.value.state() == RestoreInProgress.State.FAILURE) {
            hasFailed = true;
        }
    }
    if (hasFailed) {
        return RestoreInProgress.State.FAILURE;
    } else {
        return RestoreInProgress.State.SUCCESS;
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
private void fillLocationsFromShardIterators(Map<String, Map<String, List<Integer>>> locations,
                                             GroupShardsIterator shardIterators,
                                             List<ShardId> missingShards) {
    ShardRouting shardRouting;
    for (ShardIterator shardIterator : shardIterators) {
        shardRouting = shardIterator.nextOrNull();
        if (shardRouting != null) {
            if (shardRouting.active()) {
                processShardRouting(locations, shardRouting);
            } else {
                missingShards.add(shardIterator.shardId());
            }
        } else {
            if (isPartitioned) {
                // if the table is partitioned maybe a index/shard just got newly created ...
                missingShards.add(shardIterator.shardId());
            } else {
                throw new UnavailableShardsException(shardIterator.shardId());
            }
        }
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
/**
 * Construct a request for starting a peer recovery.
 *
 * @param shardId           the shard ID to recover
 * @param sourceNode        the source node to remover from
 * @param targetNode        the target node to recover to
 * @param metadataSnapshot  the Lucene metadata
 * @param primaryRelocation whether or not the recovery is a primary relocation
 * @param recoveryId        the recovery ID
 * @param startingSeqNo     the starting sequence number
 */
public StartRecoveryRequest(final ShardId shardId,
                            final DiscoveryNode sourceNode,
                            final DiscoveryNode targetNode,
                            final Store.MetadataSnapshot metadataSnapshot,
                            final boolean primaryRelocation,
                            final long recoveryId,
                            final long startingSeqNo) {
    this.recoveryId = recoveryId;
    this.shardId = shardId;
    this.sourceNode = sourceNode;
    this.targetNode = targetNode;
    this.metadataSnapshot = metadataSnapshot;
    this.primaryRelocation = primaryRelocation;
    this.startingSeqNo = startingSeqNo;
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
/**
 * Creates a new TranslogConfig instance
 * @param shardId the shard ID this translog belongs to
 * @param translogPath the path to use for the transaction log files
 * @param indexSettings the index settings used to set internal variables
 * @param durabilty the default durability setting for the translog
 * @param bigArrays a bigArrays instance used for temporarily allocating write operations
 * @param threadPool a {@link ThreadPool} to schedule async sync durability
 */
public TranslogConfig(ShardId shardId, Path translogPath, Settings indexSettings, Translog.Durabilty durabilty, BigArrays bigArrays, @Nullable ThreadPool threadPool) {
    this.indexSettings = indexSettings;
    this.shardId = shardId;
    this.translogPath = translogPath;
    this.durabilty = durabilty;
    this.threadPool = threadPool;
    this.bigArrays = bigArrays;
    this.type = TranslogWriter.Type.fromString(indexSettings.get(INDEX_TRANSLOG_FS_TYPE, TranslogWriter.Type.BUFFERED.name()));
    this.bufferSize = (int) indexSettings.getAsBytesSize(INDEX_TRANSLOG_BUFFER_SIZE, IndexingMemoryController.INACTIVE_SHARD_TRANSLOG_BUFFER).bytes(); // Not really interesting, updated by IndexingMemoryController...

    syncInterval = indexSettings.getAsTime(INDEX_TRANSLOG_SYNC_INTERVAL, TimeValue.timeValueSeconds(5));
    if (syncInterval.millis() > 0 && threadPool != null) {
        syncOnEachOperation = false;
    } else if (syncInterval.millis() == 0) {
        syncOnEachOperation = true;
    } else {
        syncOnEachOperation = false;
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
public void beforeIndexShardCreated(ShardId shardId, Settings indexSettings) {
    for (Listener listener : listeners) {
        try {
            listener.beforeIndexShardCreated(shardId, indexSettings);
        } catch (Throwable t) {
            logger.warn("{} failed to invoke before shard created callback", t, shardId);
            throw t;
        }
    }
}
 

import org.elasticsearch.index.shard.ShardId; //导入依赖的package包/类
public void testGetLocationWithSpaces() {
    final DocWriteResponse response =
            new DocWriteResponse(
                    new ShardId("index", "uuid", 0),
                    "type",
                    "a b",
                    SequenceNumbersService.UNASSIGNED_SEQ_NO,
                    0,
                    Result.CREATED) {};
    assertEquals("/index/type/a+b", response.getLocation(null));
    assertEquals("/index/type/a+b?routing=c+d", response.getLocation("c d"));
}