gianm commented on code in PR #17653:
URL: https://github.com/apache/druid/pull/17653#discussion_r1947039361
##########
services/src/main/java/org/apache/druid/cli/CliOverlord.java:
##########
@@ -228,6 +230,10 @@ public void configure(Binder binder)
JsonConfigProvider.bind(binder, "druid.indexer.task.default",
DefaultTaskConfig.class);
binder.bind(RetryPolicyFactory.class).in(LazySingleton.class);
+ binder.bind(SegmentMetadataCache.class)
Review Comment:
Is this meant to override the binding to `NoopSegmentMetadataCache` in
`SQLMetadataStorageDruidModule`? I thought multiple bindings typically weren't
allowed.
I wonder why we need the binding outside of CliOverlord at all- why do other
server types need to create one?
##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -453,7 +561,12 @@ public int markSegmentsUnused(final String dataSource,
final Interval interval)
*
* @return Number of segments updated.
*/
- public int markSegmentsUnused(final String dataSource, final Interval
interval, @Nullable final List<String> versions)
+ public int markSegmentsUnused(
Review Comment:
Include javadoc about what `updateTime` means. It isn't obvious from the
method signature.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ @LifecycleStop
+ public void stop()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ pollExecutor.shutdownNow();
+ datasourceToSegmentCache.forEach((datasource, cache) -> cache.stop());
+ datasourceToSegmentCache.clear();
+
+ currentCacheState = CacheState.STOPPED;
+ log.info("Stopped sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public void becomeLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ if (currentCacheState == CacheState.STOPPED) {
+ throw DruidException.defensive("Cache has not been started yet");
+ }
+
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_PENDING;
+ log.info("We are now leader. Waiting to sync latest updates from
metadata store.");
+ }
+ }
+ }
+
+ @Override
+ public void stopBeingLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ currentCacheState = CacheState.FOLLOWER;
+ log.info("Not leader anymore. Cache is now in state[%s].",
currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public boolean isEnabled()
+ {
+ return isCacheEnabled;
+ }
+
+ @Override
+ public DatasourceSegmentCache getDatasource(String dataSource)
+ {
+ verifyCacheIsReady();
+ return getCacheForDatasource(dataSource);
+ }
+
+ private HeapMemoryDatasourceSegmentCache getCacheForDatasource(String
dataSource)
+ {
+ return datasourceToSegmentCache.computeIfAbsent(dataSource,
HeapMemoryDatasourceSegmentCache::new);
+ }
+
+ /**
+ * Verifies that the cache is ready to serve requests, waiting if necessary.
+ *
+ * @throws DruidException if the cache is disabled, stopped or not leader.
+ */
+ private void verifyCacheIsReady()
+ {
+ if (!isCacheEnabled) {
+ throw DruidException.defensive("Segment metadata cache is not enabled.");
+ }
+
+ synchronized (cacheStateLock) {
+ switch (currentCacheState) {
+ case STOPPED:
+ throw DruidException.defensive("Segment metadata cache has not been
started yet.");
+ case FOLLOWER:
+ throw DruidException.defensive("Not leader yet. Segment metadata
cache is not usable.");
+ case LEADER_FIRST_SYNC_PENDING:
+ case LEADER_FIRST_SYNC_STARTED:
+ waitForCacheToFinishSync();
+ verifyCacheIsReady();
+ case LEADER_READY:
+ // Cache is now ready for use
+ }
+ }
+ }
+
+ /**
+ * Waits for cache to become ready if we are leader and current state is
+ * {@link CacheState#LEADER_FIRST_SYNC_PENDING} or
+ * {@link CacheState#LEADER_FIRST_SYNC_STARTED}.
+ */
+ private void waitForCacheToFinishSync()
+ {
+ synchronized (cacheStateLock) {
+ log.info("Waiting for cache to finish sync with metadata store.");
+ while (currentCacheState == CacheState.LEADER_FIRST_SYNC_PENDING
+ || currentCacheState == CacheState.LEADER_FIRST_SYNC_STARTED) {
+ try {
+ cacheStateLock.wait(5 * 60_000);
Review Comment:
use a static constant please. Btw, if the intent here is to have a specific
timeout on waiting for sync, it should be checked again after `wait` returns
(and then continue to `wait` if the timeout hasn't been reached yet). It is
possible for `wait` to return early in case of spurious wakeup.
To ensure the thread wakes up timely, all of the cache state transitions
should include a `notifyAll`.
##########
server/src/main/java/org/apache/druid/metadata/SegmentsMetadataManagerConfig.java:
##########
@@ -19,25 +19,42 @@
package org.apache.druid.metadata;
+import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
import org.joda.time.Period;
/**
+ * Config that dictates polling and caching of segment metadata on leader
+ * Coordinator or Overlord services.
*/
public class SegmentsMetadataManagerConfig
{
public static final String CONFIG_PREFIX = "druid.manager.segments";
@JsonProperty
- private Period pollDuration = new Period("PT1M");
+ private final Period pollDuration;
- public Period getPollDuration()
+ @JsonProperty
+ private final boolean useCache;
+
+ @JsonCreator
+ public SegmentsMetadataManagerConfig(
+ @JsonProperty("pollDuration") Period pollDuration,
+ @JsonProperty("useCache") Boolean useCache
+ )
{
- return pollDuration;
+ this.pollDuration = Configs.valueOrDefault(pollDuration,
Period.minutes(1));
+ this.useCache = Configs.valueOrDefault(useCache, true);
Review Comment:
If this is what controls whether the new caching feature is enabled, then
want this to be `false` by default for now. We could change the default to
`true` when it's more proven out.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/DruidOverlord.java:
##########
@@ -139,13 +141,15 @@ public void becomeLeader()
@Override
public void start()
{
+ segmentMetadataCache.becomeLeader();
segmentAllocationQueue.becomeLeader();
taskMaster.becomeHalfLeader(taskRunner, taskQueue);
}
@Override
public void stop()
{
+ segmentMetadataCache.stopBeingLeader();
Review Comment:
Generally, the order of items in `stop()` should be the reverse of the order
in `start()`, in case there are dependencies.
##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -442,9 +550,9 @@ public int markSegments(final Collection<SegmentId>
segmentIds, final boolean us
*
* @return Number of segments updated.
*/
- public int markSegmentsUnused(final String dataSource, final Interval
interval)
+ public int markSegmentsUnused(final String dataSource, final Interval
interval, final DateTime updateTime)
Review Comment:
Include javadoc about what `updateTime` means. It isn't obvious from the
method signature.
##########
server/src/main/java/org/apache/druid/metadata/segment/SegmentMetadataTransactionFactory.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment;
+
+/**
+ * Factory for {@link SegmentMetadataTransaction}s.
+ */
+public interface SegmentMetadataTransactionFactory
+{
+ /**
+ * Creates and executes a new read-only transaction for the given datasource.
Review Comment:
Does this method never retry? If not, why not? Seems like retries for reads
can be just as useful as retries for writes.
##########
server/src/main/java/org/apache/druid/metadata/segment/SegmentMetadataTransactionFactory.java:
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment;
+
+/**
+ * Factory for {@link SegmentMetadataTransaction}s.
+ */
+public interface SegmentMetadataTransactionFactory
+{
+ /**
+ * Creates and executes a new read-only transaction for the given datasource.
+ */
+ <T> T inReadOnlyDatasourceTransaction(
+ String dataSource,
+ SegmentMetadataReadTransaction.Callback<T> callback
+ );
+
+ /**
+ * Creates and executes a new read-write transaction for the given
datasource.
+ * The implementation may retry the transaction until it succeeds.
+ */
+ <T> T retryDatasourceTransaction(
Review Comment:
IMO, the name would be better if it was more symmetric with
`inReadOnlyDatasourceTransaction`. With the current names it sounds like
there's two ways to run a transaction:
1. read only
2. with retry
But the more important difference is "read only" vs "read-write". Maybe call
this `inReadWriteDatasourceTransaction`?
##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -149,6 +153,82 @@ public CloseableIterator<DataSegment> retrieveUsedSegments(
);
}
+ public CloseableIterator<DataSegmentPlus> retrieveUsedSegmentsPlus(
+ String dataSource,
+ Collection<Interval> intervals
+ )
+ {
+ return retrieveSegmentsPlus(
+ dataSource,
+ intervals, null, IntervalMode.OVERLAPS, true, null, null, null, null
+ );
+ }
+
+ /**
+ * Determines the highest ID amongst unused segments for the given
datasource,
+ * interval and version.
+ *
+ * @return null if no unused segment exists for the given parameters.
+ */
+ @Nullable
+ public SegmentId retrieveHighestUnusedSegmentId(
+ String datasource,
+ Interval interval,
+ String version
+ )
+ {
+ final Set<String> unusedSegmentIds =
+ retrieveUnusedSegmentIdsForExactIntervalAndVersion(datasource,
interval, version);
+ log.debug(
+ "Found [%,d] unused segments for datasource[%s] for interval[%s] and
version[%s].",
+ unusedSegmentIds.size(), datasource, interval, version
+ );
+
+ SegmentId unusedMaxId = null;
+ int maxPartitionNum = -1;
+ for (String id : unusedSegmentIds) {
+ final SegmentId segmentId = SegmentId.tryParse(datasource, id);
+ if (segmentId == null) {
Review Comment:
Warn (or perhaps even throw?) if the segment ID is unparseable, since in
that case, the method may not be returning the correct answer.
##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -264,51 +348,64 @@ public Set<SegmentId> retrieveUsedSegmentIds(
);
}
- return connector.inReadOnlyTransaction(
- (handle, status) -> {
- final Query<Map<String, Object>> sql = handle
- .createQuery(StringUtils.format(sb.toString(),
dbTables.getSegmentsTable()))
- .setFetchSize(connector.getStreamingFetchSize())
- .bind("used", true)
- .bind("dataSource", dataSource);
+ final Query<Map<String, Object>> sql = handle
+ .createQuery(StringUtils.format(sb.toString(),
dbTables.getSegmentsTable()))
+ .setFetchSize(connector.getStreamingFetchSize())
+ .bind("used", true)
+ .bind("dataSource", dataSource);
- if (compareAsString) {
- bindIntervalsToQuery(sql, Collections.singletonList(interval));
- }
+ if (compareAsString) {
+ bindIntervalsToQuery(sql, Collections.singletonList(interval));
+ }
+
+ final Set<SegmentId> segmentIds = new HashSet<>();
+ try (final ResultIterator<String> iterator =
sql.map(StringMapper.FIRST).iterator()) {
+ while (iterator.hasNext()) {
+ final String id = iterator.next();
+ final SegmentId segmentId = SegmentId.tryParse(dataSource, id);
+ if (segmentId == null) {
+ throw DruidException.defensive(
+ "Failed to parse SegmentId for id[%s] and dataSource[%s].",
+ id, dataSource
+ );
+ }
+ if (IntervalMode.OVERLAPS.apply(interval, segmentId.getInterval())) {
+ segmentIds.add(segmentId);
+ }
+ }
+ }
+ return segmentIds;
- final Set<SegmentId> segmentIds = new HashSet<>();
- try (final ResultIterator<String> iterator = sql.map((index, r, ctx)
-> r.getString(1)).iterator()) {
- while (iterator.hasNext()) {
- final String id = iterator.next();
- final SegmentId segmentId = SegmentId.tryParse(dataSource, id);
- if (segmentId == null) {
- throw DruidException.defensive(
- "Failed to parse SegmentId for id[%s] and dataSource[%s].",
- id, dataSource
- );
- }
- if (IntervalMode.OVERLAPS.apply(interval,
segmentId.getInterval())) {
- segmentIds.add(segmentId);
- }
- }
- }
- return segmentIds;
- });
}
public List<DataSegmentPlus> retrieveSegmentsById(
String datasource,
Set<String> segmentIds
)
+ {
+ try (CloseableIterator<DataSegmentPlus> iterator
+ = retrieveSegmentsByIdIterator(datasource, segmentIds)) {
+ return ImmutableList.copyOf(iterator);
+ }
+ catch (IOException e) {
+ throw DruidException.defensive(e, "Error while retrieving segments from
metadata store");
+ }
+ }
+
+ public CloseableIterator<DataSegmentPlus> retrieveSegmentsByIdIterator(
+ String datasource,
+ Set<String> segmentIds
+ )
{
final List<List<String>> partitionedSegmentIds
= Lists.partition(new ArrayList<>(segmentIds), 100);
- final List<DataSegmentPlus> fetchedSegments = new
ArrayList<>(segmentIds.size());
+ final List<CloseableIterator<DataSegmentPlus>> fetchedSegments
+ = new ArrayList<>(partitionedSegmentIds.size());
for (List<String> partition : partitionedSegmentIds) {
- fetchedSegments.addAll(retrieveSegmentBatchById(datasource, partition,
false));
+ fetchedSegments.add(retrieveSegmentBatchById(datasource, partition,
false));
Review Comment:
I wonder what this code will do exactly. There will be multiple
`CloseableIterator` from `retrieveSegmentBatchById` existing at once. What
effect does that have?
Does the metadata query get made lazily when the iterator first has
`hasNext()` called? If so then it would lead to the metadata queries being
issued sequentially, which seems fine. But, if the query is issued as part of
iterator creation, this would lead to quite a lot of simultaneously open
queries, which might cause problems with the metadata store.
##########
server/src/main/java/org/apache/druid/metadata/segment/SqlSegmentMetadataTransactionFactory.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.client.indexing.IndexingService;
+import org.apache.druid.discovery.DruidLeaderSelector;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.segment.cache.DatasourceSegmentCache;
+import org.apache.druid.metadata.segment.cache.SegmentMetadataCache;
+import org.skife.jdbi.v2.Handle;
+import org.skife.jdbi.v2.TransactionStatus;
+
+/**
+ * Factory for {@link SegmentMetadataTransaction}s. If the
+ * {@link SegmentMetadataCache} is enabled and ready, the transaction may
+ * read/write from the cache as applicable.
+ * <p>
+ * This class serves as a wrapper over the {@link SQLMetadataConnector} to
+ * perform transactions specific to segment metadata.
+ */
+public class SqlSegmentMetadataTransactionFactory implements
SegmentMetadataTransactionFactory
+{
+ private static final int QUIET_RETRIES = 3;
+ private static final int MAX_RETRIES = 10;
+
+ private final ObjectMapper jsonMapper;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+ private final DruidLeaderSelector leaderSelector;
+ private final SegmentMetadataCache segmentMetadataCache;
+
+ @Inject
+ public SqlSegmentMetadataTransactionFactory(
+ ObjectMapper jsonMapper,
+ MetadataStorageTablesConfig tablesConfig,
+ SQLMetadataConnector connector,
+ @IndexingService DruidLeaderSelector leaderSelector,
+ SegmentMetadataCache segmentMetadataCache
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.tablesConfig = tablesConfig;
+ this.connector = connector;
+ this.leaderSelector = leaderSelector;
+ this.segmentMetadataCache = segmentMetadataCache;
+ }
+
+ public int getMaxRetries()
+ {
+ return MAX_RETRIES;
+ }
+
+ @Override
+ public <T> T inReadOnlyDatasourceTransaction(
+ String dataSource,
+ SegmentMetadataReadTransaction.Callback<T> callback
+ )
+ {
+ return connector.inReadOnlyTransaction((handle, status) -> {
+ final SegmentMetadataTransaction sqlTransaction
+ = createSqlTransaction(dataSource, handle, status);
+
+ if (segmentMetadataCache.isEnabled()) {
+ final DatasourceSegmentCache datasourceCache
+ = segmentMetadataCache.getDatasource(dataSource);
+ final SegmentMetadataReadTransaction cachedTransaction
+ = new CachedSegmentMetadataTransaction(sqlTransaction,
datasourceCache, leaderSelector);
+
+ return datasourceCache.read(() ->
executeReadAndClose(cachedTransaction, callback));
+ } else {
+ return executeReadAndClose(createSqlTransaction(dataSource, handle,
status), callback);
+ }
+ });
+ }
+
+ @Override
+ public <T> T retryDatasourceTransaction(
+ String dataSource,
+ SegmentMetadataTransaction.Callback<T> callback
+ )
+ {
+ return connector.retryTransaction(
+ (handle, status) -> {
+ final SegmentMetadataTransaction sqlTransaction
+ = createSqlTransaction(dataSource, handle, status);
+
+ if (segmentMetadataCache.isEnabled()) {
+ final DatasourceSegmentCache datasourceCache
+ = segmentMetadataCache.getDatasource(dataSource);
+ final SegmentMetadataTransaction cachedTransaction
+ = new CachedSegmentMetadataTransaction(sqlTransaction,
datasourceCache, leaderSelector);
+
+ return datasourceCache.write(() ->
executeWriteAndClose(cachedTransaction, callback));
+ } else {
+ return executeWriteAndClose(sqlTransaction, callback);
+ }
+ },
+ QUIET_RETRIES,
+ getMaxRetries()
+ );
+ }
+
+ private SegmentMetadataTransaction createSqlTransaction(
+ String dataSource,
+ Handle handle,
+ TransactionStatus transactionStatus
+ )
+ {
+ return new SqlSegmentMetadataTransaction(
+ dataSource,
+ handle, transactionStatus, connector, tablesConfig, jsonMapper
+ );
+ }
+
+ private <T> T executeWriteAndClose(
+ SegmentMetadataTransaction transaction,
+ SegmentMetadataTransaction.Callback<T> callback
+ ) throws Exception
+ {
+ try {
+ return callback.inTransaction(transaction);
+ }
+ catch (Exception e) {
Review Comment:
better to catch `Throwable` for cleanup-on-error logic. We still want this
logic to fire if there is a link error, out of memory error, etc.
##########
server/src/main/java/org/apache/druid/metadata/segment/CachedSegmentMetadataTransaction.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment;
+
+import org.apache.druid.discovery.DruidLeaderSelector;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InternalServerError;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.segment.cache.DatasourceSegmentCache;
+import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.Handle;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * A {@link SegmentMetadataTransaction} that reads only from the cache and
sends
+ * writes to the metadata store. If the transaction succeeds, all the writes
+ * made to the metadata store are also committed to the cache in {@link
#close()}.
+ * The cache is not updated right away in case the transaction needs to be
+ * rolled back. This is okay since we assume that a transaction does not read
+ * what it writes.
+ */
+class CachedSegmentMetadataTransaction implements SegmentMetadataTransaction
+{
+ private final SegmentMetadataTransaction delegate;
+ private final DatasourceSegmentCache metadataCache;
+ private final DruidLeaderSelector leaderSelector;
+
+ private final int startTerm;
+
+ private final AtomicBoolean isRollingBack = new AtomicBoolean(false);
+ private final AtomicBoolean isClosed = new AtomicBoolean(false);
+
+ private final List<Consumer<DatasourceSegmentMetadataWriter>>
pendingCacheWrites = new ArrayList<>();
+
+ CachedSegmentMetadataTransaction(
+ SegmentMetadataTransaction delegate,
+ DatasourceSegmentCache metadataCache,
+ DruidLeaderSelector leaderSelector
+ )
+ {
+ this.delegate = delegate;
+ this.metadataCache = metadataCache;
+ this.leaderSelector = leaderSelector;
+
+ if (leaderSelector.isLeader()) {
+ this.startTerm = leaderSelector.localTerm();
+ } else {
+ throw InternalServerError.exception("Not leader anymore. Cannot start
transaction.");
Review Comment:
Service unavailable is a better error code. Generally we use that for any
"I'm not longer the leader" stuff, to encourage clients to retry. (Clients
don't always retry HTTP 500, but they do generally retry HTTP 503.) See
`OverlordResource#asLeaderWith` for example.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/ReadWriteCache.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.google.common.base.Supplier;
+import org.apache.druid.error.DruidException;
+
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * Cache with standard read/write locking.
+ */
+public abstract class ReadWriteCache implements DatasourceSegmentCache
+{
+ private final ReentrantReadWriteLock stateLock;
+ private volatile boolean isStopped = false;
+
+ public ReadWriteCache(boolean fair)
+ {
+ stateLock = new ReentrantReadWriteLock(fair);
+ }
+
+ /**
+ * Stops this cache. Any subsequent read/write action performed on this cache
+ * will throw a defensive DruidException.
+ */
+ public void stop()
+ {
+ withWriteLock(() -> {
+ isStopped = true;
+ });
+ }
+
+ public void withWriteLock(Action action)
+ {
+ withWriteLock(() -> {
+ action.perform();
+ return 0;
+ });
+ }
+
+ public <T> T withWriteLock(Supplier<T> action)
+ {
+ stateLock.writeLock().lock();
+ try {
+ verifyCacheIsNotStopped();
+ return action.get();
+ }
+ finally {
+ stateLock.writeLock().unlock();
+ }
+ }
+
+ public <T> T withReadLock(Supplier<T> action)
+ {
+ stateLock.readLock().lock();
+ try {
+ verifyCacheIsNotStopped();
+ return action.get();
+ }
+ finally {
+ stateLock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public <T> T read(DatasourceSegmentCache.Action<T> action) throws Exception
+ {
+ stateLock.readLock().lock();
+ try {
+ verifyCacheIsNotStopped();
+ return action.perform();
+ }
+ finally {
+ stateLock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public <T> T write(DatasourceSegmentCache.Action<T> action) throws Exception
+ {
+ stateLock.writeLock().lock();
+ try {
+ verifyCacheIsNotStopped();
+ return action.perform();
+ }
+ finally {
+ stateLock.writeLock().unlock();
+ }
+ }
+
+ private void verifyCacheIsNotStopped()
+ {
+ if (isStopped) {
+ throw DruidException.defensive("Cache is already stopped");
Review Comment:
If this happens, does it indicate a bug? Or could this happen legitimately
during e.g. server shutdown?
If it can happen legitimately, "runtime failure" is better than "defensive".
##########
server/src/main/java/org/apache/druid/metadata/segment/CachedSegmentMetadataTransaction.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment;
+
+import org.apache.druid.discovery.DruidLeaderSelector;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InternalServerError;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.segment.cache.DatasourceSegmentCache;
+import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.Handle;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * A {@link SegmentMetadataTransaction} that reads only from the cache and
sends
+ * writes to the metadata store. If the transaction succeeds, all the writes
+ * made to the metadata store are also committed to the cache in {@link
#close()}.
+ * The cache is not updated right away in case the transaction needs to be
+ * rolled back. This is okay since we assume that a transaction does not read
+ * what it writes.
+ */
+class CachedSegmentMetadataTransaction implements SegmentMetadataTransaction
+{
+ private final SegmentMetadataTransaction delegate;
+ private final DatasourceSegmentCache metadataCache;
+ private final DruidLeaderSelector leaderSelector;
+
+ private final int startTerm;
+
+ private final AtomicBoolean isRollingBack = new AtomicBoolean(false);
+ private final AtomicBoolean isClosed = new AtomicBoolean(false);
Review Comment:
minor/style preference: IMO, for state tracking, an `AtomicReference` of a
single `enum` is generally easier to follow than two `AtomicBoolean`. The code
also ends up being more "atomic".
Btw, does this code really need to be "atomic"? Do transactions need to be
thread-safe? I would think no, in which case regular booleans are preferred.
##########
server/src/main/java/org/apache/druid/metadata/segment/CachedSegmentMetadataTransaction.java:
##########
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment;
+
+import org.apache.druid.discovery.DruidLeaderSelector;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InternalServerError;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.segment.cache.DatasourceSegmentCache;
+import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.Handle;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+/**
+ * A {@link SegmentMetadataTransaction} that reads only from the cache and
sends
+ * writes to the metadata store. If the transaction succeeds, all the writes
+ * made to the metadata store are also committed to the cache in {@link
#close()}.
+ * The cache is not updated right away in case the transaction needs to be
+ * rolled back. This is okay since we assume that a transaction does not read
+ * what it writes.
+ */
+class CachedSegmentMetadataTransaction implements SegmentMetadataTransaction
+{
+ private final SegmentMetadataTransaction delegate;
+ private final DatasourceSegmentCache metadataCache;
+ private final DruidLeaderSelector leaderSelector;
+
+ private final int startTerm;
+
+ private final AtomicBoolean isRollingBack = new AtomicBoolean(false);
+ private final AtomicBoolean isClosed = new AtomicBoolean(false);
+
+ private final List<Consumer<DatasourceSegmentMetadataWriter>>
pendingCacheWrites = new ArrayList<>();
+
+ CachedSegmentMetadataTransaction(
+ SegmentMetadataTransaction delegate,
+ DatasourceSegmentCache metadataCache,
+ DruidLeaderSelector leaderSelector
+ )
+ {
+ this.delegate = delegate;
+ this.metadataCache = metadataCache;
+ this.leaderSelector = leaderSelector;
+
+ if (leaderSelector.isLeader()) {
+ this.startTerm = leaderSelector.localTerm();
+ } else {
+ throw InternalServerError.exception("Not leader anymore. Cannot start
transaction.");
+ }
+ }
+
+ private void verifyStillLeaderWithSameTerm()
+ {
+ if (!isLeaderWithSameTerm()) {
+ throw InternalServerError.exception("Not leader anymore. Failing
transaction.");
+ }
+ }
+
+ private boolean isLeaderWithSameTerm()
+ {
+ return leaderSelector.isLeader() && startTerm ==
leaderSelector.localTerm();
+ }
+
+ @Override
+ public Handle getHandle()
+ {
+ return delegate.getHandle();
+ }
+
+ @Override
+ public void setRollbackOnly()
+ {
+ isRollingBack.set(true);
+ delegate.setRollbackOnly();
+ }
+
+ @Override
+ public void close()
+ {
+ if (isClosed.get()) {
+ return;
+ } else if (isRollingBack.get()) {
+ isClosed.set(true);
+ return;
+ }
+
+ // Commit the changes to the cache
+ try {
+ pendingCacheWrites.forEach(action -> {
+ if (isLeaderWithSameTerm()) {
+ action.accept(metadataCache);
+ } else {
+ // Leadership has been lost, cache would have been stopped and
invalidated
Review Comment:
Why doesn't this throw? It seems like not-throwing would potentially lead to
callers thinking a transaction committed when it actually didn't.
##########
server/src/main/java/org/apache/druid/metadata/segment/SqlSegmentMetadataTransaction.java:
##########
@@ -0,0 +1,574 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InternalServerError;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.segment.SegmentUtils;
+import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.Handle;
+import org.skife.jdbi.v2.PreparedBatch;
+import org.skife.jdbi.v2.PreparedBatchPart;
+import org.skife.jdbi.v2.TransactionStatus;
+import org.skife.jdbi.v2.Update;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Implementation of {@link SegmentMetadataTransaction} that reads from and
+ * writes to the SQL-based metadata store directly.
+ */
+class SqlSegmentMetadataTransaction implements SegmentMetadataTransaction
+{
+ private static final int MAX_SEGMENTS_PER_BATCH = 100;
+
+ private final String dataSource;
+ private final Handle handle;
+ private final TransactionStatus transactionStatus;
+ private final SQLMetadataConnector connector;
+ private final MetadataStorageTablesConfig dbTables;
+ private final ObjectMapper jsonMapper;
+
+ private final SqlSegmentsMetadataQuery query;
+
+ SqlSegmentMetadataTransaction(
Review Comment:
Are the queries in this file all moved from other places without edits? Or
were some of them edited?
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
Review Comment:
no need for `volatile` if this is always accessed within `synchronized
(cacheStateLock)`.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
Review Comment:
suggestion: add a method like `transitionState(CacheState, String,
Object...)` and have it:
- set `currentCacheState`
- call `cacheStateLock.notifyAll`
- log the provided message, the old state, and the new state.
this way we always handle state transitions consistently.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ @LifecycleStop
+ public void stop()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ pollExecutor.shutdownNow();
+ datasourceToSegmentCache.forEach((datasource, cache) -> cache.stop());
+ datasourceToSegmentCache.clear();
+
+ currentCacheState = CacheState.STOPPED;
+ log.info("Stopped sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public void becomeLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ if (currentCacheState == CacheState.STOPPED) {
+ throw DruidException.defensive("Cache has not been started yet");
+ }
+
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_PENDING;
+ log.info("We are now leader. Waiting to sync latest updates from
metadata store.");
+ }
+ }
+ }
+
+ @Override
+ public void stopBeingLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ currentCacheState = CacheState.FOLLOWER;
+ log.info("Not leader anymore. Cache is now in state[%s].",
currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public boolean isEnabled()
+ {
+ return isCacheEnabled;
+ }
+
+ @Override
+ public DatasourceSegmentCache getDatasource(String dataSource)
+ {
+ verifyCacheIsReady();
+ return getCacheForDatasource(dataSource);
+ }
+
+ private HeapMemoryDatasourceSegmentCache getCacheForDatasource(String
dataSource)
+ {
+ return datasourceToSegmentCache.computeIfAbsent(dataSource,
HeapMemoryDatasourceSegmentCache::new);
+ }
+
+ /**
+ * Verifies that the cache is ready to serve requests, waiting if necessary.
+ *
+ * @throws DruidException if the cache is disabled, stopped or not leader.
+ */
+ private void verifyCacheIsReady()
+ {
+ if (!isCacheEnabled) {
+ throw DruidException.defensive("Segment metadata cache is not enabled.");
+ }
+
+ synchronized (cacheStateLock) {
+ switch (currentCacheState) {
+ case STOPPED:
+ throw DruidException.defensive("Segment metadata cache has not been
started yet.");
+ case FOLLOWER:
+ throw DruidException.defensive("Not leader yet. Segment metadata
cache is not usable.");
Review Comment:
If this happens, does it indicate a bug? Or could this happen legitimately
during e.g. loss of leadership?
If it can happen legitimately, "runtime failure" is better than "defensive".
Defensive should always mean "bug".
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ @LifecycleStop
+ public void stop()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ pollExecutor.shutdownNow();
+ datasourceToSegmentCache.forEach((datasource, cache) -> cache.stop());
+ datasourceToSegmentCache.clear();
+
+ currentCacheState = CacheState.STOPPED;
+ log.info("Stopped sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public void becomeLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ if (currentCacheState == CacheState.STOPPED) {
+ throw DruidException.defensive("Cache has not been started yet");
+ }
+
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_PENDING;
+ log.info("We are now leader. Waiting to sync latest updates from
metadata store.");
+ }
+ }
+ }
+
+ @Override
+ public void stopBeingLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ currentCacheState = CacheState.FOLLOWER;
+ log.info("Not leader anymore. Cache is now in state[%s].",
currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public boolean isEnabled()
+ {
+ return isCacheEnabled;
+ }
+
+ @Override
+ public DatasourceSegmentCache getDatasource(String dataSource)
+ {
+ verifyCacheIsReady();
+ return getCacheForDatasource(dataSource);
+ }
+
+ private HeapMemoryDatasourceSegmentCache getCacheForDatasource(String
dataSource)
+ {
+ return datasourceToSegmentCache.computeIfAbsent(dataSource,
HeapMemoryDatasourceSegmentCache::new);
+ }
+
+ /**
+ * Verifies that the cache is ready to serve requests, waiting if necessary.
+ *
+ * @throws DruidException if the cache is disabled, stopped or not leader.
+ */
+ private void verifyCacheIsReady()
+ {
+ if (!isCacheEnabled) {
+ throw DruidException.defensive("Segment metadata cache is not enabled.");
+ }
+
+ synchronized (cacheStateLock) {
+ switch (currentCacheState) {
+ case STOPPED:
+ throw DruidException.defensive("Segment metadata cache has not been
started yet.");
+ case FOLLOWER:
+ throw DruidException.defensive("Not leader yet. Segment metadata
cache is not usable.");
+ case LEADER_FIRST_SYNC_PENDING:
+ case LEADER_FIRST_SYNC_STARTED:
+ waitForCacheToFinishSync();
+ verifyCacheIsReady();
+ case LEADER_READY:
+ // Cache is now ready for use
+ }
+ }
+ }
+
+ /**
+ * Waits for cache to become ready if we are leader and current state is
+ * {@link CacheState#LEADER_FIRST_SYNC_PENDING} or
+ * {@link CacheState#LEADER_FIRST_SYNC_STARTED}.
+ */
+ private void waitForCacheToFinishSync()
+ {
+ synchronized (cacheStateLock) {
+ log.info("Waiting for cache to finish sync with metadata store.");
+ while (currentCacheState == CacheState.LEADER_FIRST_SYNC_PENDING
+ || currentCacheState == CacheState.LEADER_FIRST_SYNC_STARTED) {
+ try {
+ cacheStateLock.wait(5 * 60_000);
+ }
+ catch (Exception e) {
+ log.noStackTrace().error(e, "Error while waiting for cache to be
ready");
Review Comment:
I think the only error `wait` will throw during normal operation is
`InterruptedException`. Consider special-casing that, and logging at a lower
level. Other exception types can continue to be logged at error level and
re-thrown.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ @LifecycleStop
+ public void stop()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ pollExecutor.shutdownNow();
+ datasourceToSegmentCache.forEach((datasource, cache) -> cache.stop());
+ datasourceToSegmentCache.clear();
+
+ currentCacheState = CacheState.STOPPED;
+ log.info("Stopped sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public void becomeLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ if (currentCacheState == CacheState.STOPPED) {
+ throw DruidException.defensive("Cache has not been started yet");
+ }
+
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_PENDING;
+ log.info("We are now leader. Waiting to sync latest updates from
metadata store.");
+ }
+ }
+ }
+
+ @Override
+ public void stopBeingLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ currentCacheState = CacheState.FOLLOWER;
+ log.info("Not leader anymore. Cache is now in state[%s].",
currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public boolean isEnabled()
+ {
+ return isCacheEnabled;
+ }
+
+ @Override
+ public DatasourceSegmentCache getDatasource(String dataSource)
+ {
+ verifyCacheIsReady();
+ return getCacheForDatasource(dataSource);
+ }
+
+ private HeapMemoryDatasourceSegmentCache getCacheForDatasource(String
dataSource)
+ {
+ return datasourceToSegmentCache.computeIfAbsent(dataSource,
HeapMemoryDatasourceSegmentCache::new);
+ }
+
+ /**
+ * Verifies that the cache is ready to serve requests, waiting if necessary.
+ *
+ * @throws DruidException if the cache is disabled, stopped or not leader.
+ */
+ private void verifyCacheIsReady()
Review Comment:
minor suggestion: `awaitCacheReady()` would be a clearer name imo.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ @LifecycleStop
+ public void stop()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ pollExecutor.shutdownNow();
+ datasourceToSegmentCache.forEach((datasource, cache) -> cache.stop());
+ datasourceToSegmentCache.clear();
+
+ currentCacheState = CacheState.STOPPED;
+ log.info("Stopped sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public void becomeLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ if (currentCacheState == CacheState.STOPPED) {
+ throw DruidException.defensive("Cache has not been started yet");
+ }
+
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_PENDING;
Review Comment:
Interrupt and re-start the current sync? That could help the leader gain
leadership faster.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ @LifecycleStop
+ public void stop()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ pollExecutor.shutdownNow();
+ datasourceToSegmentCache.forEach((datasource, cache) -> cache.stop());
+ datasourceToSegmentCache.clear();
+
+ currentCacheState = CacheState.STOPPED;
+ log.info("Stopped sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public void becomeLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ if (currentCacheState == CacheState.STOPPED) {
+ throw DruidException.defensive("Cache has not been started yet");
+ }
+
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_PENDING;
+ log.info("We are now leader. Waiting to sync latest updates from
metadata store.");
+ }
+ }
+ }
+
+ @Override
+ public void stopBeingLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ currentCacheState = CacheState.FOLLOWER;
+ log.info("Not leader anymore. Cache is now in state[%s].",
currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public boolean isEnabled()
+ {
+ return isCacheEnabled;
+ }
+
+ @Override
+ public DatasourceSegmentCache getDatasource(String dataSource)
+ {
+ verifyCacheIsReady();
+ return getCacheForDatasource(dataSource);
+ }
+
+ private HeapMemoryDatasourceSegmentCache getCacheForDatasource(String
dataSource)
+ {
+ return datasourceToSegmentCache.computeIfAbsent(dataSource,
HeapMemoryDatasourceSegmentCache::new);
+ }
+
+ /**
+ * Verifies that the cache is ready to serve requests, waiting if necessary.
+ *
+ * @throws DruidException if the cache is disabled, stopped or not leader.
+ */
+ private void verifyCacheIsReady()
+ {
+ if (!isCacheEnabled) {
+ throw DruidException.defensive("Segment metadata cache is not enabled.");
+ }
+
+ synchronized (cacheStateLock) {
+ switch (currentCacheState) {
+ case STOPPED:
+ throw DruidException.defensive("Segment metadata cache has not been
started yet.");
+ case FOLLOWER:
+ throw DruidException.defensive("Not leader yet. Segment metadata
cache is not usable.");
+ case LEADER_FIRST_SYNC_PENDING:
+ case LEADER_FIRST_SYNC_STARTED:
+ waitForCacheToFinishSync();
+ verifyCacheIsReady();
+ case LEADER_READY:
+ // Cache is now ready for use
+ }
+ }
+ }
+
+ /**
+ * Waits for cache to become ready if we are leader and current state is
+ * {@link CacheState#LEADER_FIRST_SYNC_PENDING} or
+ * {@link CacheState#LEADER_FIRST_SYNC_STARTED}.
+ */
+ private void waitForCacheToFinishSync()
+ {
+ synchronized (cacheStateLock) {
+ log.info("Waiting for cache to finish sync with metadata store.");
+ while (currentCacheState == CacheState.LEADER_FIRST_SYNC_PENDING
+ || currentCacheState == CacheState.LEADER_FIRST_SYNC_STARTED) {
+ try {
+ cacheStateLock.wait(5 * 60_000);
+ }
+ catch (Exception e) {
+ log.noStackTrace().error(e, "Error while waiting for cache to be
ready");
+ throw DruidException.defensive(e, "Error while waiting for cache to
be ready");
+ }
+ }
+ log.info("Wait complete. Cache is now in state[%s].", currentCacheState);
+ }
+ }
+
+ private void markCacheAsReadyIfLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (currentCacheState == CacheState.LEADER_FIRST_SYNC_STARTED) {
+ currentCacheState = CacheState.LEADER_READY;
+ log.info("Sync has finished. Cache is now ready to serve requests.");
+
+ // Notify waiting threads waiting for cache to be ready
+ cacheStateLock.notifyAll();
+ }
+ }
+ }
+
+ /**
+ * Retrieves segments from the metadata store and updates the cache, if
required.
+ * <p>
+ * The following actions are performed in every sync:
+ * <ul>
+ * <li>Retrieve all used and unused segment IDs along with their updated
timestamps</li>
+ * <li>Retrieve payloads of used segments which have been updated in the
metadata
+ * store but not in the cache</li>
+ * <li>Retrieve all pending segments and update the cache as needed</li>
+ * <li>Remove segments not present in the metadata store</li>
+ * <li>Reset the max unused partition IDs</li>
+ * <li>Change the cache state to ready if it is leader and waiting for first
sync</li>
+ * <li>Emit metrics</li>
+ * </ul>
+ */
+ private void syncWithMetadataStore()
+ {
+ final DateTime pollStartTime = DateTimes.nowUtc();
+ final Stopwatch sincePollStart = Stopwatch.createStarted();
+ try {
+ synchronized (cacheStateLock) {
+ if (currentCacheState == CacheState.LEADER_FIRST_SYNC_PENDING) {
+ log.info("Started sync of latest updates from metadata store.");
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_STARTED;
+ }
+ }
+
+ final Map<String, DatasourceSegmentSummary> datasourceToSummary = new
HashMap<>();
+ retrieveAllSegmentIds(datasourceToSummary);
+
+ datasourceToSegmentCache.keySet().forEach(
+ dataSource -> removeUnknownSegmentsFromCache(
+ dataSource,
+ datasourceToSummary.computeIfAbsent(dataSource, ds -> new
DatasourceSegmentSummary()),
+ pollStartTime
+ )
+ );
+
+ datasourceToSummary.forEach(this::retrieveAndRefreshUsedSegments);
+
+ retrieveAndRefreshAllPendingSegments(datasourceToSummary);
+ datasourceToSegmentCache.keySet().forEach(
+ dataSource -> removeUnknownPendingSegmentsFromCache(
+ dataSource,
+ datasourceToSummary.computeIfAbsent(dataSource, ds -> new
DatasourceSegmentSummary()),
+ pollStartTime
+ )
+ );
+
+ datasourceToSegmentCache.values().forEach(
+ HeapMemoryDatasourceSegmentCache::markCacheSynced
+ );
+
+ datasourceToSummary.forEach(this::emitSummaryMetrics);
+
+ final long pollDurationMillis = sincePollStart.millisElapsed();
+ emitMetric("sync/time", pollDurationMillis);
+ syncFinishTime.set(DateTimes.nowUtc());
+
+ markCacheAsReadyIfLeader();
+ }
+ catch (Throwable t) {
+ log.error(t, "Error occurred while polling metadata store");
+ log.makeAlert(t, "Error occurred while polling metadata store");
Review Comment:
Missing call to `emit()` after `makeAlert`. Also, no reason to call both
`log.error` and `log.makeAlert`. The alert is logged when it is emitted.
##########
server/src/main/java/org/apache/druid/metadata/segment/cache/HeapMemorySegmentMetadataCache.java:
##########
@@ -0,0 +1,623 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata.segment.cache;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Supplier;
+import com.google.errorprone.annotations.ThreadSafe;
+import com.google.errorprone.annotations.concurrent.GuardedBy;
+import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Stopwatch;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
+import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.MetadataStorageTablesConfig;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.metadata.SQLMetadataConnector;
+import org.apache.druid.metadata.SegmentsMetadataManagerConfig;
+import org.apache.druid.metadata.SqlSegmentsMetadataQuery;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.server.http.DataSegmentPlus;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.DateTime;
+import org.joda.time.Duration;
+import org.joda.time.Interval;
+import org.skife.jdbi.v2.ResultIterator;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * In-memory implementation of {@link SegmentMetadataCache}.
+ */
+@ThreadSafe
+public class HeapMemorySegmentMetadataCache implements SegmentMetadataCache
+{
+ private static final EmittingLogger log = new
EmittingLogger(HeapMemorySegmentMetadataCache.class);
+ private static final String METRIC_PREFIX = "segment/metadataCache/";
+
+ private enum CacheState
+ {
+ STOPPED, FOLLOWER, LEADER_FIRST_SYNC_PENDING, LEADER_FIRST_SYNC_STARTED,
LEADER_READY
+ }
+
+ private final ObjectMapper jsonMapper;
+ private final Duration pollDuration;
+ private final boolean isCacheEnabled;
+ private final MetadataStorageTablesConfig tablesConfig;
+ private final SQLMetadataConnector connector;
+
+ private final ScheduledExecutorService pollExecutor;
+ private final ServiceEmitter emitter;
+
+ private final Object cacheStateLock = new Object();
+
+ @GuardedBy("cacheStateLock")
+ private volatile CacheState currentCacheState = CacheState.STOPPED;
+
+ private final ConcurrentHashMap<String, HeapMemoryDatasourceSegmentCache>
+ datasourceToSegmentCache = new ConcurrentHashMap<>();
+ private final AtomicReference<DateTime> syncFinishTime = new
AtomicReference<>();
+
+ @Inject
+ public HeapMemorySegmentMetadataCache(
+ ObjectMapper jsonMapper,
+ Supplier<SegmentsMetadataManagerConfig> config,
+ Supplier<MetadataStorageTablesConfig> tablesConfig,
+ SQLMetadataConnector connector,
+ ScheduledExecutorFactory executorFactory,
+ ServiceEmitter emitter
+ )
+ {
+ this.jsonMapper = jsonMapper;
+ this.isCacheEnabled = config.get().isUseCache();
+ this.pollDuration = config.get().getPollDuration().toStandardDuration();
+ this.tablesConfig = tablesConfig.get();
+ this.connector = connector;
+ this.pollExecutor = isCacheEnabled ? executorFactory.create(1,
"SegmentMetadataCache-%s") : null;
+ this.emitter = emitter;
+ }
+
+
+ @Override
+ @LifecycleStart
+ public void start()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled && currentCacheState == CacheState.STOPPED) {
+ currentCacheState = CacheState.FOLLOWER;
+ pollExecutor.schedule(this::syncWithMetadataStore,
pollDuration.getMillis(), TimeUnit.MILLISECONDS);
+
+ log.info("Starting sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ @LifecycleStop
+ public void stop()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ pollExecutor.shutdownNow();
+ datasourceToSegmentCache.forEach((datasource, cache) -> cache.stop());
+ datasourceToSegmentCache.clear();
+
+ currentCacheState = CacheState.STOPPED;
+ log.info("Stopped sync with metadata store. Cache is now in
state[%s].", currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public void becomeLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ if (currentCacheState == CacheState.STOPPED) {
+ throw DruidException.defensive("Cache has not been started yet");
+ }
+
+ currentCacheState = CacheState.LEADER_FIRST_SYNC_PENDING;
+ log.info("We are now leader. Waiting to sync latest updates from
metadata store.");
+ }
+ }
+ }
+
+ @Override
+ public void stopBeingLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (isCacheEnabled) {
+ currentCacheState = CacheState.FOLLOWER;
+ log.info("Not leader anymore. Cache is now in state[%s].",
currentCacheState);
+ }
+ }
+ }
+
+ @Override
+ public boolean isEnabled()
+ {
+ return isCacheEnabled;
+ }
+
+ @Override
+ public DatasourceSegmentCache getDatasource(String dataSource)
+ {
+ verifyCacheIsReady();
+ return getCacheForDatasource(dataSource);
+ }
+
+ private HeapMemoryDatasourceSegmentCache getCacheForDatasource(String
dataSource)
+ {
+ return datasourceToSegmentCache.computeIfAbsent(dataSource,
HeapMemoryDatasourceSegmentCache::new);
+ }
+
+ /**
+ * Verifies that the cache is ready to serve requests, waiting if necessary.
+ *
+ * @throws DruidException if the cache is disabled, stopped or not leader.
+ */
+ private void verifyCacheIsReady()
+ {
+ if (!isCacheEnabled) {
+ throw DruidException.defensive("Segment metadata cache is not enabled.");
+ }
+
+ synchronized (cacheStateLock) {
+ switch (currentCacheState) {
+ case STOPPED:
+ throw DruidException.defensive("Segment metadata cache has not been
started yet.");
+ case FOLLOWER:
+ throw DruidException.defensive("Not leader yet. Segment metadata
cache is not usable.");
+ case LEADER_FIRST_SYNC_PENDING:
+ case LEADER_FIRST_SYNC_STARTED:
+ waitForCacheToFinishSync();
+ verifyCacheIsReady();
+ case LEADER_READY:
+ // Cache is now ready for use
+ }
+ }
+ }
+
+ /**
+ * Waits for cache to become ready if we are leader and current state is
+ * {@link CacheState#LEADER_FIRST_SYNC_PENDING} or
+ * {@link CacheState#LEADER_FIRST_SYNC_STARTED}.
+ */
+ private void waitForCacheToFinishSync()
+ {
+ synchronized (cacheStateLock) {
+ log.info("Waiting for cache to finish sync with metadata store.");
+ while (currentCacheState == CacheState.LEADER_FIRST_SYNC_PENDING
+ || currentCacheState == CacheState.LEADER_FIRST_SYNC_STARTED) {
+ try {
+ cacheStateLock.wait(5 * 60_000);
+ }
+ catch (Exception e) {
+ log.noStackTrace().error(e, "Error while waiting for cache to be
ready");
+ throw DruidException.defensive(e, "Error while waiting for cache to
be ready");
+ }
+ }
+ log.info("Wait complete. Cache is now in state[%s].", currentCacheState);
+ }
+ }
+
+ private void markCacheAsReadyIfLeader()
+ {
+ synchronized (cacheStateLock) {
+ if (currentCacheState == CacheState.LEADER_FIRST_SYNC_STARTED) {
+ currentCacheState = CacheState.LEADER_READY;
+ log.info("Sync has finished. Cache is now ready to serve requests.");
+
+ // Notify waiting threads waiting for cache to be ready
+ cacheStateLock.notifyAll();
+ }
+ }
+ }
+
+ /**
+ * Retrieves segments from the metadata store and updates the cache, if
required.
+ * <p>
+ * The following actions are performed in every sync:
+ * <ul>
+ * <li>Retrieve all used and unused segment IDs along with their updated
timestamps</li>
+ * <li>Retrieve payloads of used segments which have been updated in the
metadata
+ * store but not in the cache</li>
+ * <li>Retrieve all pending segments and update the cache as needed</li>
+ * <li>Remove segments not present in the metadata store</li>
+ * <li>Reset the max unused partition IDs</li>
+ * <li>Change the cache state to ready if it is leader and waiting for first
sync</li>
+ * <li>Emit metrics</li>
+ * </ul>
+ */
+ private void syncWithMetadataStore()
Review Comment:
Have you been able to benchmark this method with a cluster with lots of
segments (e.g. millions)? It will need to complete before allocation can work
post-leadership-election, so I'm wondering how much time that will take.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
