org.apache.lucene.search.QueryCachingPolicy#org.elasticsearch.index.translog.TranslogConfig源码实例Demo

public LocalTranslog(TranslogConfig config) throws IOException {
    super(config.getShardId(), config.getIndexSettings());
    ReadWriteLock rwl = new ReentrantReadWriteLock();
    readLock = new ReleasableLock(rwl.readLock());
    writeLock = new ReleasableLock(rwl.writeLock());
    this.translogPath = config.getTranslogPath();
    // clean all files
    Files.createDirectories(this.translogPath);
    Files.walkFileTree(this.translogPath, new SimpleFileVisitor<Path>() {
        @Override
        public FileVisitResult visitFile(Path file,
                BasicFileAttributes attrs) throws IOException {
            Files.delete(file);
            return FileVisitResult.CONTINUE;
        }
    });
    
    // create a new directory
    writeChannel = FileChannel.open(this.translogPath.resolve(getFileNameFromId(tmpTranslogGeneration.get())), StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
    writtenOffset = 0;
}
 

private final EngineConfig newEngineConfig(TranslogConfig translogConfig, QueryCachingPolicy cachingPolicy) {
    final TranslogRecoveryPerformer translogRecoveryPerformer = new TranslogRecoveryPerformer(shardId, mapperService,
            queryParserService,
            indexAliasesService, indexCache, logger) {
        @Override
        protected void operationProcessed() {
            assert recoveryState != null;
            recoveryState.getTranslog().incrementRecoveredOperations();
        }

        @Override
        public int recoveryFromSnapshot(Engine engine, Translog.Snapshot snapshot) throws IOException {
            assert recoveryState != null;
            RecoveryState.Translog translogStats = recoveryState.getTranslog();
            translogStats.totalOperations(snapshot.estimatedTotalOperations());
            translogStats.totalOperationsOnStart(snapshot.estimatedTotalOperations());
            return super.recoveryFromSnapshot(engine, snapshot);
        }
    };
    return new EngineConfig(shardId,
            threadPool, indexingService, indexSettingsService.indexSettings(), warmer, store, deletionPolicy, mergePolicyConfig
            .getMergePolicy(), mergeSchedulerConfig,
            mapperService.indexAnalyzer(), similarityService.similarity(), codecService, failedEngineListener,
            translogRecoveryPerformer, indexCache.query(), cachingPolicy, wrappingService, translogConfig);
}
 

private Translog openTranslog(EngineConfig engineConfig, TranslogDeletionPolicy translogDeletionPolicy,
                                    LongSupplier globalCheckpointSupplier) throws IOException {

    final TranslogConfig translogConfig = engineConfig.getTranslogConfig();
    final String translogUUID = loadTranslogUUIDFromLastCommit();
    // We expect that this shard already exists, so it must already have an existing translog else something is badly wrong!
    return new Translog(translogConfig, translogUUID, translogDeletionPolicy, globalCheckpointSupplier,
        engineConfig.getPrimaryTermSupplier());
}
 

public DistributedTranslog(TranslogConfig config, String nodeId) throws IOException {
    super(config, nodeId);
    this.translogConfig = config;
    this.localNodeId = nodeId;
    String indexUUID = this.indexSettings.get(IndexMetaData.SETTING_INDEX_UUID, IndexMetaData.INDEX_UUID_NA_VALUE);
    this.logName = getLogName(config.getShardId().getIndex(), indexUUID, this.shardId.getId());
    this.numOperations = 0;
    this.sizeInBytes = 0;
    this.prepareNumOperations = 0;
    this.prepareSizeInBytes = 0;
    this.localTranslog = new LocalTranslog(config);
}
 

/**
 * Creates a new {@link org.elasticsearch.index.engine.EngineConfig}
 */
public EngineConfig(ShardId shardId, ThreadPool threadPool, ShardIndexingService indexingService,
                    Settings indexSettings, IndicesWarmer warmer, Store store, SnapshotDeletionPolicy deletionPolicy,
                    MergePolicy mergePolicy, MergeSchedulerConfig mergeSchedulerConfig, Analyzer analyzer,
                    Similarity similarity, CodecService codecService, Engine.FailedEngineListener failedEngineListener,
                    TranslogRecoveryPerformer translogRecoveryPerformer, QueryCache queryCache, QueryCachingPolicy queryCachingPolicy, IndexSearcherWrappingService wrappingService, TranslogConfig translogConfig) {
    this.shardId = shardId;
    this.indexSettings = indexSettings;
    this.threadPool = threadPool;
    this.indexingService = indexingService;
    this.warmer = warmer;
    this.store = store;
    this.deletionPolicy = deletionPolicy;
    this.mergePolicy = mergePolicy;
    this.mergeSchedulerConfig = mergeSchedulerConfig;
    this.analyzer = analyzer;
    this.similarity = similarity;
    this.codecService = codecService;
    this.failedEngineListener = failedEngineListener;
    this.wrappingService = wrappingService;
    this.optimizeAutoGenerateId = indexSettings.getAsBoolean(EngineConfig.INDEX_OPTIMIZE_AUTOGENERATED_ID_SETTING, false);
    this.compoundOnFlush = indexSettings.getAsBoolean(EngineConfig.INDEX_COMPOUND_ON_FLUSH, compoundOnFlush);
    codecName = indexSettings.get(EngineConfig.INDEX_CODEC_SETTING, EngineConfig.DEFAULT_CODEC_NAME);
    // We start up inactive and rely on IndexingMemoryController to give us our fair share once we start indexing:
    indexingBufferSize = IndexingMemoryController.INACTIVE_SHARD_INDEXING_BUFFER;
    gcDeletesInMillis = indexSettings.getAsTime(INDEX_GC_DELETES_SETTING, EngineConfig.DEFAULT_GC_DELETES).millis();
    versionMapSizeSetting = indexSettings.get(INDEX_VERSION_MAP_SIZE, DEFAULT_VERSION_MAP_SIZE);
    updateVersionMapSize();
    this.translogRecoveryPerformer = translogRecoveryPerformer;
    this.forceNewTranslog = indexSettings.getAsBoolean(INDEX_FORCE_NEW_TRANSLOG, false);
    this.queryCache = queryCache;
    this.queryCachingPolicy = queryCachingPolicy;
    this.translogConfig = translogConfig;
}
 

private Translog openTranslog(EngineConfig engineConfig, IndexWriter writer, boolean createNew) throws IOException {
    final Translog.TranslogGeneration generation = loadTranslogIdFromCommit(writer);
    final TranslogConfig translogConfig = engineConfig.getTranslogConfig();

    if (createNew == false) {
        // We expect that this shard already exists, so it must already have an existing translog else something is badly wrong!
        if (generation == null) {
            throw new IllegalStateException("no translog generation present in commit data but translog is expected to exist");
        }
        translogConfig.setTranslogGeneration(generation);
        if (generation != null && generation.translogUUID == null) {
            // only upgrade on pre-2.0 indices...
            Translog.upgradeLegacyTranslog(logger, translogConfig);
        }
    }
    final Translog translog = new Translog(translogConfig);
    if (generation == null || generation.translogUUID == null) {
        if (generation == null) {
            logger.debug("no translog ID present in the current generation - creating one");
        } else if (generation.translogUUID == null) {
            logger.debug("upgraded translog to pre 2.0 format, associating translog with index - writing translog UUID");
        }
        boolean success = false;
        try {
            commitIndexWriter(writer, translog);
            success = true;
        } finally {
            if (success == false) {
                IOUtils.closeWhileHandlingException(translog);
            }
        }
    }
    return translog;
}
 

private static Translog.Durabilty getFromSettings(ESLogger logger, Settings settings, Translog.Durabilty defaultValue) {
    final String value = settings.get(TranslogConfig.INDEX_TRANSLOG_DURABILITY, defaultValue.name());
    try {
        return Translog.Durabilty.valueOf(value.toUpperCase(Locale.ROOT));
    } catch (IllegalArgumentException ex) {
        logger.warn("Can't apply {} illegal value: {} using {} instead, use one of: {}", TranslogConfig.INDEX_TRANSLOG_DURABILITY,
                value, defaultValue, Arrays.toString(Translog.Durabilty.values()));
        return defaultValue;
    }
}
 

private Translog openTranslog(EngineConfig engineConfig, TranslogDeletionPolicy translogDeletionPolicy,
                              LongSupplier globalCheckpointSupplier, LongConsumer persistedSequenceNumberConsumer) throws IOException {

    final TranslogConfig translogConfig = engineConfig.getTranslogConfig();
    final String translogUUID = loadTranslogUUIDFromLastCommit();
    // We expect that this shard already exists, so it must already have an existing translog else something is badly wrong!
    return new Translog(translogConfig, translogUUID, translogDeletionPolicy, globalCheckpointSupplier,
        engineConfig.getPrimaryTermSupplier(), persistedSequenceNumberConsumer);
}
 

protected Translog createTranslog(Path translogPath, LongSupplier primaryTermSupplier) throws IOException {
    TranslogConfig translogConfig = new TranslogConfig(shardId, translogPath, INDEX_SETTINGS, BigArrays.NON_RECYCLING_INSTANCE);
    String translogUUID = Translog.createEmptyTranslog(translogPath, SequenceNumbers.NO_OPS_PERFORMED, shardId,
        primaryTermSupplier.getAsLong());
    return new Translog(translogConfig, translogUUID, createTranslogDeletionPolicy(INDEX_SETTINGS),
        () -> SequenceNumbers.NO_OPS_PERFORMED, primaryTermSupplier, seqNo -> {});
}
 

/**
 * Returns the translog config for this engine
 */
public TranslogConfig getTranslogConfig() {
    return translogConfig;
}
 

private void registerBuiltinIndexSettings() {
    registerIndexDynamicSetting(IndexStore.INDEX_STORE_THROTTLE_MAX_BYTES_PER_SEC, Validator.BYTES_SIZE);
    registerIndexDynamicSetting(IndexStore.INDEX_STORE_THROTTLE_TYPE, Validator.EMPTY);
    registerIndexDynamicSetting(MergeSchedulerConfig.MAX_THREAD_COUNT, Validator.EMPTY);
    registerIndexDynamicSetting(MergeSchedulerConfig.MAX_MERGE_COUNT, Validator.EMPTY);
    registerIndexDynamicSetting(MergeSchedulerConfig.AUTO_THROTTLE, Validator.EMPTY);
    registerIndexDynamicSetting(FilterAllocationDecider.INDEX_ROUTING_REQUIRE_GROUP + "*", Validator.EMPTY);
    registerIndexDynamicSetting(FilterAllocationDecider.INDEX_ROUTING_INCLUDE_GROUP + "*", Validator.EMPTY);
    registerIndexDynamicSetting(FilterAllocationDecider.INDEX_ROUTING_EXCLUDE_GROUP + "*", Validator.EMPTY);
    registerIndexDynamicSetting(EnableAllocationDecider.INDEX_ROUTING_ALLOCATION_ENABLE, Validator.EMPTY);
    registerIndexDynamicSetting(EnableAllocationDecider.INDEX_ROUTING_REBALANCE_ENABLE, Validator.EMPTY);
    registerIndexDynamicSetting(DisableAllocationDecider.INDEX_ROUTING_ALLOCATION_DISABLE_ALLOCATION, Validator.EMPTY);
    registerIndexDynamicSetting(DisableAllocationDecider.INDEX_ROUTING_ALLOCATION_DISABLE_NEW_ALLOCATION, Validator.EMPTY);
    registerIndexDynamicSetting(DisableAllocationDecider.INDEX_ROUTING_ALLOCATION_DISABLE_REPLICA_ALLOCATION, Validator.EMPTY);
    registerIndexDynamicSetting(TranslogConfig.INDEX_TRANSLOG_FS_TYPE, Validator.EMPTY);
    registerIndexDynamicSetting(IndexMetaData.SETTING_NUMBER_OF_REPLICAS, Validator.NON_NEGATIVE_INTEGER);
    registerIndexDynamicSetting(IndexMetaData.SETTING_AUTO_EXPAND_REPLICAS, Validator.EMPTY);
    registerIndexDynamicSetting(IndexMetaData.SETTING_READ_ONLY, Validator.EMPTY);
    registerIndexDynamicSetting(IndexMetaData.SETTING_BLOCKS_READ, Validator.EMPTY);
    registerIndexDynamicSetting(IndexMetaData.SETTING_BLOCKS_WRITE, Validator.EMPTY);
    registerIndexDynamicSetting(IndexMetaData.SETTING_BLOCKS_METADATA, Validator.EMPTY);
    registerIndexDynamicSetting(IndexMetaData.SETTING_SHARED_FS_ALLOW_RECOVERY_ON_ANY_NODE, Validator.EMPTY);
    registerIndexDynamicSetting(IndexMetaData.SETTING_PRIORITY, Validator.NON_NEGATIVE_INTEGER);
    registerIndexDynamicSetting(IndicesTTLService.INDEX_TTL_DISABLE_PURGE, Validator.EMPTY);
    registerIndexDynamicSetting(IndexShard.INDEX_REFRESH_INTERVAL, Validator.TIME);
    registerIndexDynamicSetting(IndexShard.INDEX_REINDEX_INTERVAL, Validator.TIME);
    registerIndexDynamicSetting(IndexShard.INDEX_REINDEX_BATCH_SIZE, Validator.NON_NEGATIVE_INTEGER);
    registerIndexDynamicSetting(PrimaryShardAllocator.INDEX_RECOVERY_INITIAL_SHARDS, Validator.EMPTY);
    registerIndexDynamicSetting(EngineConfig.INDEX_COMPOUND_ON_FLUSH, Validator.BOOLEAN);
    registerIndexDynamicSetting(EngineConfig.INDEX_GC_DELETES_SETTING, Validator.TIME);
    registerIndexDynamicSetting(IndexShard.INDEX_FLUSH_ON_CLOSE, Validator.BOOLEAN);
    registerIndexDynamicSetting(EngineConfig.INDEX_VERSION_MAP_SIZE, Validator.BYTES_SIZE_OR_PERCENTAGE);
    registerIndexDynamicSetting(IndexingSlowLog.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_WARN, Validator.TIME);
    registerIndexDynamicSetting(IndexingSlowLog.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_INFO, Validator.TIME);
    registerIndexDynamicSetting(IndexingSlowLog.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_DEBUG, Validator.TIME);
    registerIndexDynamicSetting(IndexingSlowLog.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_TRACE, Validator.TIME);
    registerIndexDynamicSetting(IndexingSlowLog.INDEX_INDEXING_SLOWLOG_REFORMAT, Validator.EMPTY);
    registerIndexDynamicSetting(IndexingSlowLog.INDEX_INDEXING_SLOWLOG_LEVEL, Validator.EMPTY);
    registerIndexDynamicSetting(IndexingSlowLog.INDEX_INDEXING_SLOWLOG_MAX_SOURCE_CHARS_TO_LOG, Validator.EMPTY);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_WARN, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_INFO, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_DEBUG, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_TRACE, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_WARN, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_INFO, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_DEBUG, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_TRACE, Validator.TIME);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_REFORMAT, Validator.EMPTY);
    registerIndexDynamicSetting(SearchSlowLog.INDEX_SEARCH_SLOWLOG_LEVEL, Validator.EMPTY);
    registerIndexDynamicSetting(ShardsLimitAllocationDecider.INDEX_TOTAL_SHARDS_PER_NODE, Validator.INTEGER);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED, Validator.DOUBLE);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_MERGE_POLICY_FLOOR_SEGMENT, Validator.BYTES_SIZE);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE, Validator.INTEGER_GTE_2);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_EXPLICIT, Validator.INTEGER_GTE_2);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT, Validator.BYTES_SIZE);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER, Validator.DOUBLE_GTE_2);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_MERGE_POLICY_RECLAIM_DELETES_WEIGHT, Validator.NON_NEGATIVE_DOUBLE);
    registerIndexDynamicSetting(MergePolicyConfig.INDEX_COMPOUND_FORMAT, Validator.EMPTY);
    registerIndexDynamicSetting(TranslogService.INDEX_TRANSLOG_FLUSH_INTERVAL, Validator.TIME);
    registerIndexDynamicSetting(TranslogService.INDEX_TRANSLOG_FLUSH_THRESHOLD_OPS, Validator.INTEGER);
    registerIndexDynamicSetting(TranslogService.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE, Validator.BYTES_SIZE);
    registerIndexDynamicSetting(TranslogService.INDEX_TRANSLOG_FLUSH_THRESHOLD_PERIOD, Validator.TIME);
    registerIndexDynamicSetting(TranslogService.INDEX_TRANSLOG_DISABLE_FLUSH, Validator.EMPTY);
    registerIndexDynamicSetting(TranslogConfig.INDEX_TRANSLOG_DURABILITY, Validator.EMPTY);
    registerIndexDynamicSetting(IndicesWarmer.INDEX_WARMER_ENABLED, Validator.EMPTY);
    registerIndexDynamicSetting(IndicesRequestCache.INDEX_CACHE_REQUEST_ENABLED, Validator.BOOLEAN);
    registerIndexDynamicSetting(IndicesRequestCache.DEPRECATED_INDEX_CACHE_REQUEST_ENABLED, Validator.BOOLEAN);
    registerIndexDynamicSetting(UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING, Validator.TIME);
    registerIndexDynamicSetting(DefaultSearchContext.MAX_RESULT_WINDOW, Validator.POSITIVE_INTEGER);
    registerIndexDynamicSetting(MapperService.INDEX_MAPPING_NESTED_FIELDS_LIMIT_SETTING, Validator.NON_NEGATIVE_INTEGER);
    registerIndexDynamicSetting(IndexMetaData.SETTING_DL_SNAPSHOT_RECOVERY_ROWS, Validator.NON_NEGATIVE_INTEGER);
}
 

/**
 * Creates a new {@link org.elasticsearch.index.engine.EngineConfig}
 */
public EngineConfig(ShardId shardId,
                    String allocationId,
                    ThreadPool threadPool,
                    IndexSettings indexSettings,
                    Store store,
                    MergePolicy mergePolicy,
                    Analyzer analyzer,
                    CodecService codecService,
                    Engine.EventListener eventListener,
                    QueryCache queryCache,
                    QueryCachingPolicy queryCachingPolicy,
                    TranslogConfig translogConfig,
                    TimeValue flushMergesAfter,
                    List<ReferenceManager.RefreshListener> externalRefreshListener,
                    List<ReferenceManager.RefreshListener> internalRefreshListener,
                    CircuitBreakerService circuitBreakerService,
                    LongSupplier globalCheckpointSupplier,
                    LongSupplier primaryTermSupplier,
                    TombstoneDocSupplier tombstoneDocSupplier) {
    this.shardId = shardId;
    this.allocationId = allocationId;
    this.indexSettings = indexSettings;
    this.threadPool = threadPool;
    this.store = store;
    this.mergePolicy = mergePolicy;
    this.analyzer = analyzer;
    this.codecService = codecService;
    this.eventListener = eventListener;
    codecName = indexSettings.getValue(INDEX_CODEC_SETTING);
    // We need to make the indexing buffer for this shard at least as large
    // as the amount of memory that is available for all engines on the
    // local node so that decisions to flush segments to disk are made by
    // IndexingMemoryController rather than Lucene.
    // Add an escape hatch in case this change proves problematic - it used
    // to be a fixed amound of RAM: 256 MB.
    // TODO: Remove this escape hatch in 8.x
    final String escapeHatchProperty = "es.index.memory.max_index_buffer_size";
    String maxBufferSize = System.getProperty(escapeHatchProperty);
    if (maxBufferSize != null) {
        indexingBufferSize = MemorySizeValue.parseBytesSizeValueOrHeapRatio(maxBufferSize, escapeHatchProperty);
    } else {
        indexingBufferSize = IndexingMemoryController.INDEX_BUFFER_SIZE_SETTING.get(indexSettings.getNodeSettings());
    }
    this.queryCache = queryCache;
    this.queryCachingPolicy = queryCachingPolicy;
    this.translogConfig = translogConfig;
    this.flushMergesAfter = flushMergesAfter;
    this.externalRefreshListener = externalRefreshListener;
    this.internalRefreshListener = internalRefreshListener;
    this.circuitBreakerService = circuitBreakerService;
    this.globalCheckpointSupplier = globalCheckpointSupplier;
    this.primaryTermSupplier = primaryTermSupplier;
    this.tombstoneDocSupplier = tombstoneDocSupplier;
}
 

/**
 * Returns the translog config for this engine
 */
public TranslogConfig getTranslogConfig() {
    return translogConfig;
}
 

public EngineConfig config(IndexSettings indexSettings, Store store, Path translogPath, MergePolicy mergePolicy,
                           ReferenceManager.RefreshListener externalRefreshListener,
                           ReferenceManager.RefreshListener internalRefreshListener,
                           LongSupplier globalCheckpointSupplier) {
    IndexWriterConfig iwc = newIndexWriterConfig();
    TranslogConfig translogConfig = new TranslogConfig(shardId, translogPath, indexSettings, BigArrays.NON_RECYCLING_INSTANCE);
    Engine.EventListener listener = new Engine.EventListener() {
        @Override
        public void onFailedEngine(String reason, @Nullable Exception e) {
            // we don't need to notify anybody in this test
        }
    };
    final List<ReferenceManager.RefreshListener> extRefreshListenerList =
        externalRefreshListener == null ? emptyList() : Collections.singletonList(externalRefreshListener);
    final List<ReferenceManager.RefreshListener> intRefreshListenerList =
        internalRefreshListener == null ? emptyList() : Collections.singletonList(internalRefreshListener);

    if (globalCheckpointSupplier == null) {
        globalCheckpointSupplier = new ReplicationTracker(shardId, allocationId.getId(), indexSettings, SequenceNumbers.NO_OPS_PERFORMED, update -> {
        });
    }

    return new EngineConfig(
        shardId,
        allocationId.getId(),
        threadPool,
        indexSettings,
        store,
        mergePolicy,
        iwc.getAnalyzer(),
        new CodecService(null, logger),
        listener,
        IndexSearcher.getDefaultQueryCache(),
        IndexSearcher.getDefaultQueryCachingPolicy(),
        translogConfig,
        TimeValue.timeValueMinutes(5),
        extRefreshListenerList,
        intRefreshListenerList,
        new NoneCircuitBreakerService(),
        globalCheckpointSupplier,
        primaryTerm,
        tombstoneDocSupplier());
}