This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new a71e77c9d0a Remove code for CuratorLoadQueuePeon (#17657)
a71e77c9d0a is described below
commit a71e77c9d0a947eb1b85271c9a6779bcea60a909
Author: Kashif Faraz <[email protected]>
AuthorDate: Thu Jan 23 19:52:57 2025 +0530
Remove code for CuratorLoadQueuePeon (#17657)
ZK-based segment loading was completely disabled in #16360 and #15705.
This PR removes all related code which is now unused.
Changes:
- Remove CuratorLoadQueuePeon and CuratorLoadQueuePeonTest
- Refactor DruidCoordinatorTest to use a TestLoadQueuePeon instead
---
.../coordinator/loading/CuratorLoadQueuePeon.java | 427 --------------------
.../server/coordinator/loading/LoadQueuePeon.java | 1 -
.../coordinator/CuratorDruidCoordinatorTest.java | 439 ---------------------
.../server/coordinator/DruidCoordinatorTest.java | 311 ++++-----------
.../loading/CuratorLoadQueuePeonTest.java | 401 -------------------
.../druid/server/coordinator/simulate/README.md | 1 -
6 files changed, 67 insertions(+), 1513 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeon.java
b/server/src/main/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeon.java
deleted file mode 100644
index 0ab9d8ab329..00000000000
---
a/server/src/main/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeon.java
+++ /dev/null
@@ -1,427 +0,0 @@
-/*
- * 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.server.coordinator.loading;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.api.CuratorWatcher;
-import org.apache.curator.utils.ZKPaths;
-import org.apache.druid.java.util.common.ISE;
-import org.apache.druid.java.util.emitter.EmittingLogger;
-import org.apache.druid.server.coordination.SegmentChangeRequestNoop;
-import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
-import org.apache.druid.server.coordinator.stats.Stats;
-import org.apache.druid.timeline.DataSegment;
-import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
-import org.apache.zookeeper.data.Stat;
-import org.joda.time.Duration;
-
-import javax.annotation.Nonnull;
-import javax.annotation.Nullable;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.ConcurrentSkipListSet;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * Use {@link HttpLoadQueuePeon} instead.
- * <p>
- * Objects of this class can be accessed by multiple threads. State wise, this
class
- * is thread safe and callers of the public methods can expect thread safe
behavior.
- * Though, like a typical object being accessed by multiple threads,
- * callers shouldn't expect strict consistency in results between two calls
- * of the same or different methods.
- */
-@Deprecated
-public class CuratorLoadQueuePeon implements LoadQueuePeon
-{
- private static final EmittingLogger log = new
EmittingLogger(CuratorLoadQueuePeon.class);
-
- private final CuratorFramework curator;
- private final String basePath;
- private final ObjectMapper jsonMapper;
- private final ScheduledExecutorService processingExecutor;
-
- /**
- * Threadpool with daemon threads that execute callback actions associated
- * with loading or dropping segments.
- */
- private final ExecutorService callBackExecutor;
- private final Duration loadTimeout;
-
- private final AtomicLong queuedSize = new AtomicLong(0);
- private final AtomicReference<CoordinatorRunStats> stats = new
AtomicReference<>(new CoordinatorRunStats());
-
- /**
- * Needs to be thread safe since it can be concurrently accessed via
- * {@link #loadSegment(DataSegment, SegmentAction, LoadPeonCallback)},
- * {@link #actionCompleted(SegmentHolder)}, {@link #getSegmentsToLoad()} and
- * {@link #stop()}.
- */
- private final ConcurrentSkipListMap<DataSegment, SegmentHolder>
segmentsToLoad
- = new ConcurrentSkipListMap<>(SegmentHolder.NEWEST_SEGMENT_FIRST);
-
- /**
- * Needs to be thread safe since it can be concurrently accessed via
- * {@link #dropSegment(DataSegment, LoadPeonCallback)}, {@link
#actionCompleted(SegmentHolder)},
- * {@link #getSegmentsToDrop()} and {@link #stop()}
- */
- private final ConcurrentSkipListMap<DataSegment, SegmentHolder>
segmentsToDrop
- = new ConcurrentSkipListMap<>(SegmentHolder.NEWEST_SEGMENT_FIRST);
-
- /**
- * Needs to be thread safe since it can be concurrently accessed via
- * {@link #markSegmentToDrop(DataSegment)}}, {@link
#unmarkSegmentToDrop(DataSegment)}}
- * and {@link #getSegmentsToDrop()}
- */
- private final ConcurrentSkipListSet<DataSegment> segmentsMarkedToDrop
- = new ConcurrentSkipListSet<>(SegmentHolder.NEWEST_SEGMENT_FIRST);
-
- /**
- * Needs to be thread safe since it can be concurrently accessed via
- * {@link #failAssign(SegmentHolder, boolean, Exception)}, {@link
#actionCompleted(SegmentHolder)},
- * {@link #getTimedOutSegments()} and {@link #stop()}
- */
- private final ConcurrentSkipListSet<DataSegment> timedOutSegments =
- new ConcurrentSkipListSet<>(SegmentHolder.NEWEST_SEGMENT_FIRST);
-
- public CuratorLoadQueuePeon(
- CuratorFramework curator,
- String basePath,
- ObjectMapper jsonMapper,
- ScheduledExecutorService processingExecutor,
- ExecutorService callbackExecutor,
- Duration loadTimeout
- )
- {
- this.curator = curator;
- this.basePath = basePath;
- this.jsonMapper = jsonMapper;
- this.callBackExecutor = callbackExecutor;
- this.processingExecutor = processingExecutor;
- this.loadTimeout = loadTimeout;
- }
-
- @JsonProperty
- @Override
- public Set<DataSegment> getSegmentsToLoad()
- {
- return segmentsToLoad.keySet();
- }
-
- @Override
- public Set<SegmentHolder> getSegmentsInQueue()
- {
- final Set<SegmentHolder> segmentsInQueue = new HashSet<>();
- segmentsInQueue.addAll(segmentsToLoad.values());
- segmentsInQueue.addAll(segmentsToDrop.values());
- return segmentsInQueue;
- }
-
- @JsonProperty
- @Override
- public Set<DataSegment> getSegmentsToDrop()
- {
- return segmentsToDrop.keySet();
- }
-
- @JsonProperty
- @Override
- public Set<DataSegment> getSegmentsMarkedToDrop()
- {
- return segmentsMarkedToDrop;
- }
-
- @Override
- public Set<DataSegment> getTimedOutSegments()
- {
- return timedOutSegments;
- }
-
- @Override
- public long getSizeOfSegmentsToLoad()
- {
- return queuedSize.get();
- }
-
- @Override
- public long getLoadRateKbps()
- {
- return 0;
- }
-
- @Override
- public CoordinatorRunStats getAndResetStats()
- {
- return stats.getAndSet(new CoordinatorRunStats());
- }
-
- @Override
- public void loadSegment(final DataSegment segment, SegmentAction action,
@Nullable final LoadPeonCallback callback)
- {
- SegmentHolder segmentHolder = new SegmentHolder(segment, action,
Duration.ZERO, callback);
- final SegmentHolder existingHolder = segmentsToLoad.putIfAbsent(segment,
segmentHolder);
- if (existingHolder != null) {
- existingHolder.addCallback(callback);
- return;
- }
- log.debug("Asking server peon[%s] to load segment[%s]", basePath,
segment.getId());
- queuedSize.addAndGet(segment.getSize());
- processingExecutor.submit(new SegmentChangeProcessor(segmentHolder));
- }
-
- @Override
- public void dropSegment(final DataSegment segment, @Nullable final
LoadPeonCallback callback)
- {
- SegmentHolder segmentHolder = new SegmentHolder(segment,
SegmentAction.DROP, Duration.ZERO, callback);
- final SegmentHolder existingHolder = segmentsToDrop.putIfAbsent(segment,
segmentHolder);
- if (existingHolder != null) {
- existingHolder.addCallback(callback);
- return;
- }
- log.debug("Asking server peon[%s] to drop segment[%s]", basePath,
segment.getId());
- processingExecutor.submit(new SegmentChangeProcessor(segmentHolder));
- }
-
- @Override
- public void markSegmentToDrop(DataSegment dataSegment)
- {
- segmentsMarkedToDrop.add(dataSegment);
- }
-
- @Override
- public void unmarkSegmentToDrop(DataSegment dataSegment)
- {
- segmentsMarkedToDrop.remove(dataSegment);
- }
-
- private class SegmentChangeProcessor implements Runnable
- {
- private final SegmentHolder segmentHolder;
-
- private SegmentChangeProcessor(SegmentHolder segmentHolder)
- {
- this.segmentHolder = segmentHolder;
- }
-
- @Override
- public void run()
- {
- try {
- final String path = ZKPaths.makePath(basePath,
segmentHolder.getSegmentIdentifier());
- final byte[] payload =
jsonMapper.writeValueAsBytes(segmentHolder.getChangeRequest());
- curator.create().withMode(CreateMode.EPHEMERAL).forPath(path, payload);
- if (log.isDebugEnabled()) {
- log.debug(
- "ZKNode created for server to [%s] %s [%s]",
- basePath, segmentHolder.getAction(),
segmentHolder.getSegmentIdentifier()
- );
- }
- final ScheduledFuture<?> nodeDeletedCheck =
scheduleNodeDeletedCheck(path);
- final Stat stat = curator.checkExists().usingWatcher(
- (CuratorWatcher) watchedEvent -> {
- switch (watchedEvent.getType()) {
- case NodeDeleted:
- // Cancel the check node deleted task since we have already
- // been notified by the zk watcher
- nodeDeletedCheck.cancel(true);
- onZkNodeDeleted(segmentHolder, watchedEvent.getPath());
- break;
- default:
- // do nothing
- }
- }
- ).forPath(path);
-
- // Cleanup watcher to avoid memory leak if we missed the NodeDeleted
event
- if (stat == null) {
- // Create a node and then delete it to remove the registered
watcher. This is a work-around for
- // a zookeeper race condition. Specifically, when you set a
watcher, it fires on the next event
- // that happens for that node. If no events happen, the watcher
stays registered foreverz.
- // Couple that with the fact that you cannot set a watcher when you
create a node, but what we
- // want is to create a node and then watch for it to get deleted.
The solution is that you *can*
- // set a watcher when you check to see if it exists so, we first
create the node and then set a
- // watcher on its existence. However, if already does not exist by
the time the existence check
- // returns, then the watcher that was set will never fire (nobody
will ever create the node
- // again) and thus lead to a slow, but real, memory leak. So, we
create another node to cause
- // that watcher to fire and delete it right away.
- //
- // We do not create the existence watcher first, because then it
will fire when we create the
- // node and we'll have the same race when trying to refresh that
watcher.
- final byte[] noopPayload = jsonMapper.writeValueAsBytes(new
SegmentChangeRequestNoop());
- curator.create().withMode(CreateMode.EPHEMERAL).forPath(path,
noopPayload);
- onZkNodeDeleted(segmentHolder, path);
- }
- }
- catch (KeeperException.NodeExistsException ne) {
- // This is expected when historicals haven't yet picked up processing
this segment and coordinator
- // tries reassigning it to the same node.
- log.warn(ne, "ZK node already exists because segment change request
hasn't yet been processed");
- failAssign(segmentHolder, true, null);
- }
- catch (Exception e) {
- failAssign(segmentHolder, false, e);
- }
- }
-
- @Nonnull
- private ScheduledFuture<?> scheduleNodeDeletedCheck(String path)
- {
- return processingExecutor.schedule(
- () -> {
- try {
- if (curator.checkExists().forPath(path) != null) {
- failAssign(
- segmentHolder,
- true,
- new ISE(
- "%s operation timed out and [%s] was never removed! "
- + "These segments may still get processed.",
- segmentHolder.getAction(),
- path
- )
- );
- } else {
- log.debug("Path [%s] has been removed.", path);
- }
- }
- catch (Exception e) {
- log.error(e, "Exception caught and ignored when checking whether
zk node was deleted");
- failAssign(segmentHolder, false, e);
- }
- },
- loadTimeout.getMillis(),
- TimeUnit.MILLISECONDS
- );
- }
- }
-
- private void actionCompleted(SegmentHolder segmentHolder)
- {
- switch (segmentHolder.getAction()) {
- case LOAD:
- case REPLICATE:
- case MOVE_TO:
- // When load failed a segment will be removed from the segmentsToLoad
twice and
- // null value will be returned at the second time in which case
queueSize may be negative.
- // See https://github.com/apache/druid/pull/10362 for more details.
- if (null != segmentsToLoad.remove(segmentHolder.getSegment())) {
- queuedSize.addAndGet(-segmentHolder.getSegment().getSize());
- timedOutSegments.remove(segmentHolder.getSegment());
- }
- break;
- case DROP:
- segmentsToDrop.remove(segmentHolder.getSegment());
- timedOutSegments.remove(segmentHolder.getSegment());
- break;
- default:
- throw new UnsupportedOperationException();
- }
- executeCallbacks(segmentHolder, true);
- }
-
-
- @Override
- public void start()
- {
- }
-
- @Override
- public void stop()
- {
- for (SegmentHolder holder : segmentsToDrop.values()) {
- executeCallbacks(holder, false);
- }
- segmentsToDrop.clear();
-
- for (SegmentHolder holder : segmentsToLoad.values()) {
- executeCallbacks(holder, false);
- }
- segmentsToLoad.clear();
-
- timedOutSegments.clear();
- queuedSize.set(0L);
- stats.get().clear();
- }
-
- private void onZkNodeDeleted(SegmentHolder segmentHolder, String path)
- {
- if
(!ZKPaths.getNodeFromPath(path).equals(segmentHolder.getSegmentIdentifier())) {
- log.warn(
- "Server[%s] entry [%s] was removed even though it's not what is
currently loading[%s]",
- basePath,
- path,
- segmentHolder
- );
- return;
- }
- actionCompleted(segmentHolder);
- log.debug(
- "Server[%s] done processing %s of segment [%s]",
- basePath,
- segmentHolder.getAction(),
- path
- );
- }
-
- private void failAssign(SegmentHolder segmentHolder, boolean handleTimeout,
Exception e)
- {
- if (e != null) {
- log.error(e, "Server[%s], throwable caught when submitting [%s].",
basePath, segmentHolder);
- }
- stats.get().add(Stats.SegmentQueue.FAILED_ACTIONS, 1);
-
- if (handleTimeout) {
- // Avoid removing the segment entry from the load/drop list in case
config.getLoadTimeoutDelay() expires.
- // This is because the ZK Node is still present and it may be processed
after this timeout and so the coordinator
- // needs to take this into account.
- log.debug(
- "Skipping segment removal from [%s] queue, since ZK Node still
exists!",
- segmentHolder.getAction()
- );
- timedOutSegments.add(segmentHolder.getSegment());
- executeCallbacks(segmentHolder, false);
- } else {
- // This may have failed for a different reason and so act like it was
completed.
- actionCompleted(segmentHolder);
- }
- }
-
- @Override
- public boolean cancelOperation(DataSegment segment)
- {
- return false;
- }
-
- private void executeCallbacks(SegmentHolder holder, boolean success)
- {
- for (LoadPeonCallback callback : holder.getCallbacks()) {
- callBackExecutor.submit(() -> callback.execute(success));
- }
- }
-}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueuePeon.java
b/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueuePeon.java
index 8e9989717ab..2d42a49feba 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueuePeon.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/loading/LoadQueuePeon.java
@@ -27,7 +27,6 @@ import java.util.Set;
/**
* Supports load queue management.
*/
-@Deprecated
public interface LoadQueuePeon
{
void start();
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java
deleted file mode 100644
index c312cb493b5..00000000000
---
a/server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java
+++ /dev/null
@@ -1,439 +0,0 @@
-/*
- * 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.server.coordinator;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Predicates;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.recipes.cache.PathChildrenCache;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
-import org.apache.curator.utils.ZKPaths;
-import org.apache.druid.client.BatchServerInventoryView;
-import org.apache.druid.client.CoordinatorSegmentWatcherConfig;
-import org.apache.druid.client.CoordinatorServerView;
-import org.apache.druid.client.DataSourcesSnapshot;
-import org.apache.druid.client.DruidServer;
-import org.apache.druid.client.ImmutableDruidDataSource;
-import org.apache.druid.curator.CuratorTestBase;
-import org.apache.druid.curator.CuratorUtils;
-import org.apache.druid.jackson.DefaultObjectMapper;
-import org.apache.druid.java.util.common.Intervals;
-import org.apache.druid.java.util.common.concurrent.Execs;
-import org.apache.druid.segment.TestHelper;
-import org.apache.druid.segment.realtime.appenderator.SegmentSchemas;
-import org.apache.druid.server.coordination.DruidServerMetadata;
-import org.apache.druid.server.coordination.ServerType;
-import org.apache.druid.server.coordinator.balancer.BalancerStrategy;
-import org.apache.druid.server.coordinator.loading.CuratorLoadQueuePeon;
-import org.apache.druid.server.coordinator.loading.LoadQueuePeon;
-import org.apache.druid.server.coordinator.loading.LoadQueueTaskMaster;
-import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
-import org.apache.druid.server.coordinator.loading.SegmentLoadingConfig;
-import org.apache.druid.server.coordinator.loading.StrategicSegmentAssigner;
-import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
-import org.apache.druid.server.initialization.ZkPathsConfig;
-import org.apache.druid.server.metrics.NoopServiceEmitter;
-import org.apache.druid.testing.DeadlockDetectingTimeout;
-import org.apache.druid.timeline.DataSegment;
-import org.apache.druid.timeline.SegmentId;
-import org.apache.druid.timeline.partition.NoneShardSpec;
-import org.easymock.EasyMock;
-import org.joda.time.Duration;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TestRule;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-/**
- * This tests zookeeper specific coordinator/load queue/historical
interactions, such as moving segments by the balancer
- */
-public class CuratorDruidCoordinatorTest extends CuratorTestBase
-{
- private DataSourcesSnapshot dataSourcesSnapshot;
- private DruidCoordinatorRuntimeParams coordinatorRuntimeParams;
-
- private LoadQueuePeon sourceLoadQueuePeon;
- private LoadQueuePeon destinationLoadQueuePeon;
- private PathChildrenCache sourceLoadQueueChildrenCache;
- private PathChildrenCache destinationLoadQueueChildrenCache;
-
- private static final String SEGPATH = "/druid/segments";
- private static final String SOURCE_LOAD_PATH =
"/druid/loadQueue/localhost:1";
- private static final String DESTINATION_LOAD_PATH =
"/druid/loadQueue/localhost:2";
- private static final long COORDINATOR_START_DELAY = 1;
- private static final long COORDINATOR_PERIOD = 100;
-
- private BatchServerInventoryView baseView;
- private CoordinatorServerView serverView;
- private CountDownLatch segmentViewInitLatch;
- /**
- * The following two fields are changed during {@link #testMoveSegment()},
the change might not be visible from the
- * thread, that runs the callback, registered in {@link #setupView()}.
volatile modificator doesn't guarantee
- * visibility either, but somewhat increases the chances.
- */
- private volatile CountDownLatch segmentAddedLatch;
- private volatile CountDownLatch segmentRemovedLatch;
- private final ObjectMapper jsonMapper;
- private final ZkPathsConfig zkPathsConfig;
-
- private final ScheduledExecutorService peonExec =
Execs.scheduledSingleThreaded("Master-PeonExec--%d");
- private final ExecutorService callbackExec = Execs.multiThreaded(4,
"LoadQueuePeon-callbackexec--%d");
-
- public CuratorDruidCoordinatorTest()
- {
- jsonMapper = TestHelper.makeJsonMapper();
- zkPathsConfig = new ZkPathsConfig();
- }
-
- @Before
- public void setUp() throws Exception
- {
- dataSourcesSnapshot = EasyMock.createNiceMock(DataSourcesSnapshot.class);
- coordinatorRuntimeParams =
EasyMock.createNiceMock(DruidCoordinatorRuntimeParams.class);
-
- setupServerAndCurator();
- curator.start();
- curator.blockUntilConnected();
- curator.create().creatingParentsIfNeeded().forPath(SEGPATH);
- curator.create().creatingParentsIfNeeded().forPath(SOURCE_LOAD_PATH);
- curator.create().creatingParentsIfNeeded().forPath(DESTINATION_LOAD_PATH);
-
- final ObjectMapper objectMapper = new DefaultObjectMapper();
- sourceLoadQueueChildrenCache = new PathChildrenCache(
- curator,
- SOURCE_LOAD_PATH,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache_src-%d")
- );
- destinationLoadQueueChildrenCache = new PathChildrenCache(
- curator,
- DESTINATION_LOAD_PATH,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache_dest-%d")
- );
- sourceLoadQueuePeon = new CuratorLoadQueuePeon(
- curator,
- SOURCE_LOAD_PATH,
- objectMapper,
- peonExec,
- callbackExec,
- Duration.standardMinutes(15)
- );
- destinationLoadQueuePeon = new CuratorLoadQueuePeon(
- curator,
- DESTINATION_LOAD_PATH,
- objectMapper,
- peonExec,
- callbackExec,
- Duration.standardMinutes(15)
- );
- }
-
- @After
- public void tearDown() throws Exception
- {
- baseView.stop();
- sourceLoadQueuePeon.stop();
- sourceLoadQueueChildrenCache.close();
- destinationLoadQueueChildrenCache.close();
- tearDownServerAndCurator();
- }
-
- @Rule
- public final TestRule timeout = new DeadlockDetectingTimeout(60,
TimeUnit.SECONDS);
-
- @Test
- public void testStopDoesntKillPoolItDoesntOwn() throws Exception
- {
- setupView();
- sourceLoadQueuePeon.stop();
- Assert.assertFalse(peonExec.isShutdown());
- Assert.assertFalse(callbackExec.isShutdown());
- }
-
- @Test
- public void testMoveSegment() throws Exception
- {
- segmentViewInitLatch = new CountDownLatch(1);
- segmentAddedLatch = new CountDownLatch(4);
-
- segmentRemovedLatch = new CountDownLatch(0);
-
- setupView();
-
- DruidServer source = new DruidServer(
- "localhost:1",
- "localhost:1",
- null,
- 10000000L,
- ServerType.HISTORICAL,
- "default_tier",
- 0
- );
-
- DruidServer dest = new DruidServer(
- "localhost:2",
- "localhost:2",
- null,
- 10000000L,
- ServerType.HISTORICAL,
- "default_tier",
- 0
- );
-
- setupZNodeForServer(source, zkPathsConfig, jsonMapper);
- setupZNodeForServer(dest, zkPathsConfig, jsonMapper);
-
- final List<DataSegment> sourceSegments = Arrays.asList(
- createSegment("2011-04-01/2011-04-03", "v1"),
- createSegment("2011-04-03/2011-04-06", "v1"),
- createSegment("2011-04-06/2011-04-09", "v1")
- );
-
- final List<DataSegment> destinationSegments = Collections.singletonList(
- createSegment("2011-03-31/2011-04-01", "v1")
- );
-
- DataSegment segmentToMove = sourceSegments.get(2);
-
- List<String> sourceSegKeys = new ArrayList<>();
-
- for (DataSegment segment : sourceSegments) {
- sourceSegKeys.add(announceBatchSegmentsForServer(source,
ImmutableSet.of(segment), zkPathsConfig, jsonMapper));
- }
-
- for (DataSegment segment : destinationSegments) {
- announceBatchSegmentsForServer(dest, ImmutableSet.of(segment),
zkPathsConfig, jsonMapper);
- }
-
- Assert.assertTrue(timing.forWaiting().awaitLatch(segmentViewInitLatch));
- Assert.assertTrue(timing.forWaiting().awaitLatch(segmentAddedLatch));
-
- // these child watchers are used to simulate actions of historicals,
announcing a segment on noticing a load queue
- // for the destination and unannouncing from source server when noticing a
drop request
-
- CountDownLatch srcCountdown = new CountDownLatch(1);
- sourceLoadQueueChildrenCache.getListenable().addListener(
- (CuratorFramework curatorFramework, PathChildrenCacheEvent event) -> {
- if (event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED))
{
- srcCountdown.countDown();
- } else if (CuratorUtils.isChildAdded(event)) {
- //Simulate source server dropping segment
- unannounceSegmentFromBatchForServer(source, segmentToMove,
sourceSegKeys.get(2), zkPathsConfig);
- }
- }
- );
-
- CountDownLatch destCountdown = new CountDownLatch(1);
- destinationLoadQueueChildrenCache.getListenable().addListener(
- (CuratorFramework curatorFramework, PathChildrenCacheEvent event) -> {
- if (event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED))
{
- destCountdown.countDown();
- } else if (CuratorUtils.isChildAdded(event)) {
- //Simulate destination server loading segment
- announceBatchSegmentsForServer(dest,
ImmutableSet.of(segmentToMove), zkPathsConfig, jsonMapper);
- }
- }
- );
-
-
sourceLoadQueueChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
-
destinationLoadQueueChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
-
- Assert.assertTrue(timing.forWaiting().awaitLatch(srcCountdown));
- Assert.assertTrue(timing.forWaiting().awaitLatch(destCountdown));
-
- sourceSegments.forEach(source::addDataSegment);
- destinationSegments.forEach(dest::addDataSegment);
-
- segmentRemovedLatch = new CountDownLatch(1);
- segmentAddedLatch = new CountDownLatch(1);
-
- ImmutableDruidDataSource druidDataSource =
EasyMock.createNiceMock(ImmutableDruidDataSource.class);
-
EasyMock.expect(druidDataSource.getSegment(EasyMock.anyObject(SegmentId.class))).andReturn(sourceSegments.get(2));
- EasyMock.replay(druidDataSource);
- EasyMock.expect(coordinatorRuntimeParams.getDataSourcesSnapshot())
- .andReturn(dataSourcesSnapshot).anyTimes();
- final CoordinatorDynamicConfig dynamicConfig =
-
CoordinatorDynamicConfig.builder().withUseRoundRobinSegmentAssignment(false).build();
- EasyMock.expect(coordinatorRuntimeParams.getCoordinatorDynamicConfig())
- .andReturn(dynamicConfig)
- .anyTimes();
- EasyMock.expect(coordinatorRuntimeParams.getSegmentLoadingConfig())
- .andReturn(SegmentLoadingConfig.create(dynamicConfig, 100))
- .anyTimes();
-
- final ServerHolder sourceServer = new
ServerHolder(source.toImmutableDruidServer(), sourceLoadQueuePeon);
- final ServerHolder destinationServer = new
ServerHolder(dest.toImmutableDruidServer(), destinationLoadQueuePeon);
- final DruidCluster cluster =
DruidCluster.builder().add(sourceServer).add(destinationServer).build();
-
- final BalancerStrategy balancerStrategy =
EasyMock.mock(BalancerStrategy.class);
- EasyMock.expect(
- balancerStrategy.findDestinationServerToMoveSegment(
- EasyMock.anyObject(),
- EasyMock.anyObject(),
- EasyMock.anyObject()
- )
- ).andReturn(destinationServer).atLeastOnce();
- EasyMock.expect(coordinatorRuntimeParams.getBalancerStrategy())
- .andReturn(balancerStrategy).anyTimes();
-
EasyMock.expect(coordinatorRuntimeParams.getDruidCluster()).andReturn(cluster).anyTimes();
- EasyMock.replay(coordinatorRuntimeParams, balancerStrategy);
-
- EasyMock.expect(dataSourcesSnapshot.getDataSource(EasyMock.anyString()))
- .andReturn(druidDataSource).anyTimes();
- EasyMock.replay(dataSourcesSnapshot);
-
- LoadQueueTaskMaster taskMaster =
EasyMock.createMock(LoadQueueTaskMaster.class);
- EasyMock.expect(taskMaster.isHttpLoading()).andReturn(false).anyTimes();
- EasyMock.replay(taskMaster);
-
- // Move the segment from source to dest
- SegmentLoadQueueManager loadQueueManager =
- new SegmentLoadQueueManager(baseView, taskMaster);
- StrategicSegmentAssigner segmentAssigner =
createSegmentAssigner(loadQueueManager, coordinatorRuntimeParams);
- segmentAssigner.moveSegment(
- segmentToMove,
- sourceServer,
- Collections.singletonList(destinationServer)
- );
-
- // wait for destination server to load segment
- Assert.assertTrue(timing.forWaiting().awaitLatch(segmentAddedLatch));
-
- // remove load queue key from destination server to trigger adding drop to
load queue
-
curator.delete().guaranteed().forPath(ZKPaths.makePath(DESTINATION_LOAD_PATH,
segmentToMove.getId().toString()));
-
- // wait for drop
- Assert.assertTrue(timing.forWaiting().awaitLatch(segmentRemovedLatch));
-
- // clean up drop from load queue
- curator.delete().guaranteed().forPath(ZKPaths.makePath(SOURCE_LOAD_PATH,
segmentToMove.getId().toString()));
-
- List<DruidServer> servers = new ArrayList<>(serverView.getInventory());
-
- Assert.assertEquals(2, servers.get(0).getTotalSegments());
- Assert.assertEquals(2, servers.get(1).getTotalSegments());
- }
-
- private void setupView() throws Exception
- {
- baseView = new BatchServerInventoryView(
- zkPathsConfig,
- curator,
- jsonMapper,
- Predicates.alwaysTrue(),
- "test"
- )
- {
- @Override
- public void registerSegmentCallback(Executor exec, final SegmentCallback
callback)
- {
- super.registerSegmentCallback(
- exec,
- new SegmentCallback()
- {
- @Override
- public CallbackAction segmentAdded(DruidServerMetadata server,
DataSegment segment)
- {
- CallbackAction res = callback.segmentAdded(server, segment);
- segmentAddedLatch.countDown();
- return res;
- }
-
- @Override
- public CallbackAction segmentRemoved(DruidServerMetadata server,
DataSegment segment)
- {
- CallbackAction res = callback.segmentRemoved(server, segment);
- segmentRemovedLatch.countDown();
- return res;
- }
-
- @Override
- public CallbackAction segmentViewInitialized()
- {
- CallbackAction res = callback.segmentViewInitialized();
- segmentViewInitLatch.countDown();
- return res;
- }
-
- @Override
- public CallbackAction segmentSchemasAnnounced(SegmentSchemas
segmentSchemas)
- {
- return CallbackAction.CONTINUE;
- }
- }
- );
- }
- };
-
- serverView = new CoordinatorServerView(baseView, new
CoordinatorSegmentWatcherConfig(), new NoopServiceEmitter(), null);
-
- baseView.start();
-
- sourceLoadQueuePeon.start();
- destinationLoadQueuePeon.start();
- }
-
- private DataSegment createSegment(String intervalStr, String version)
- {
- return DataSegment.builder()
- .dataSource("test_curator_druid_coordinator")
- .interval(Intervals.of(intervalStr))
- .loadSpec(ImmutableMap.of("type", "local", "path",
"somewhere"))
- .version(version)
- .dimensions(ImmutableList.of())
- .metrics(ImmutableList.of())
- .shardSpec(NoneShardSpec.instance())
- .binaryVersion(9)
- .size(0)
- .build();
- }
-
- private StrategicSegmentAssigner createSegmentAssigner(
- SegmentLoadQueueManager loadQueueManager,
- DruidCoordinatorRuntimeParams params
- )
- {
- return new StrategicSegmentAssigner(
- loadQueueManager,
- params.getDruidCluster(),
- params.getBalancerStrategy(),
- params.getSegmentLoadingConfig(),
- new CoordinatorRunStats()
- );
- }
-}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
index 512efd0316e..d50e82e7baa 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/DruidCoordinatorTest.java
@@ -25,27 +25,22 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import it.unimi.dsi.fastutil.objects.Object2IntMap;
import it.unimi.dsi.fastutil.objects.Object2LongMap;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.recipes.cache.PathChildrenCache;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
-import org.apache.druid.client.BatchServerInventoryView;
import org.apache.druid.client.DataSourcesSnapshot;
import org.apache.druid.client.DruidDataSource;
import org.apache.druid.client.DruidServer;
import org.apache.druid.client.ImmutableDruidDataSource;
import org.apache.druid.client.ImmutableDruidServer;
+import org.apache.druid.client.ServerInventoryView;
import org.apache.druid.common.config.JacksonConfigManager;
-import org.apache.druid.curator.CuratorTestBase;
-import org.apache.druid.curator.CuratorUtils;
import org.apache.druid.curator.discovery.LatchableServiceAnnouncer;
import org.apache.druid.discovery.DruidLeaderSelector;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.Intervals;
-import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
import org.apache.druid.java.util.emitter.core.Event;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
import org.apache.druid.metadata.MetadataRuleManager;
import org.apache.druid.metadata.SegmentsMetadataManager;
import org.apache.druid.rpc.indexing.OverlordClient;
@@ -65,37 +60,37 @@ import
org.apache.druid.server.coordinator.duty.CoordinatorCustomDutyGroup;
import org.apache.druid.server.coordinator.duty.CoordinatorCustomDutyGroups;
import org.apache.druid.server.coordinator.duty.DutyGroupStatus;
import org.apache.druid.server.coordinator.duty.KillSupervisorsCustomDuty;
-import org.apache.druid.server.coordinator.loading.CuratorLoadQueuePeon;
+import org.apache.druid.server.coordinator.loading.LoadPeonCallback;
import org.apache.druid.server.coordinator.loading.LoadQueuePeon;
import org.apache.druid.server.coordinator.loading.LoadQueueTaskMaster;
+import org.apache.druid.server.coordinator.loading.SegmentAction;
import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
+import org.apache.druid.server.coordinator.loading.TestLoadQueuePeon;
import
org.apache.druid.server.coordinator.rules.ForeverBroadcastDistributionRule;
import org.apache.druid.server.coordinator.rules.ForeverLoadRule;
import org.apache.druid.server.coordinator.rules.IntervalLoadRule;
import org.apache.druid.server.coordinator.rules.Rule;
+import org.apache.druid.server.coordinator.stats.Stats;
import org.apache.druid.server.lookup.cache.LookupCoordinatorManager;
import org.apache.druid.timeline.DataSegment;
import org.easymock.EasyMock;
import org.joda.time.Duration;
-import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import javax.annotation.Nullable;
import java.util.Collections;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
/**
*/
-public class DruidCoordinatorTest extends CuratorTestBase
+public class DruidCoordinatorTest
{
- private static final String LOADPATH = "/druid/loadqueue/localhost:1234";
- private static final Duration LOAD_TIMEOUT = Duration.standardMinutes(15);
private static final long COORDINATOR_START_DELAY = 1;
private static final long COORDINATOR_PERIOD = 100;
private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
@@ -104,27 +99,22 @@ public class DruidCoordinatorTest extends CuratorTestBase
private SegmentsMetadataManager segmentsMetadataManager;
private DataSourcesSnapshot dataSourcesSnapshot;
- private BatchServerInventoryView serverInventoryView;
+ private ServerInventoryView serverInventoryView;
private ScheduledExecutorFactory scheduledExecutorFactory;
- private DruidServer druidServer;
- private LoadQueuePeon loadQueuePeon;
private LoadQueueTaskMaster loadQueueTaskMaster;
private MetadataRuleManager metadataRuleManager;
private CountDownLatch leaderAnnouncerLatch;
private CountDownLatch leaderUnannouncerLatch;
- private PathChildrenCache pathChildrenCache;
private DruidCoordinatorConfig druidCoordinatorConfig;
- private ObjectMapper objectMapper;
private DruidNode druidNode;
private OverlordClient overlordClient;
private CompactionStatusTracker statusTracker;
- private final LatchableServiceEmitter serviceEmitter = new
LatchableServiceEmitter();
+ private LatchableServiceEmitter serviceEmitter;
@Before
public void setUp() throws Exception
{
- druidServer = new DruidServer("from", "from", null, 5L,
ServerType.HISTORICAL, "tier1", 0);
- serverInventoryView = EasyMock.createMock(BatchServerInventoryView.class);
+ serverInventoryView = EasyMock.createMock(ServerInventoryView.class);
segmentsMetadataManager =
EasyMock.createNiceMock(SegmentsMetadataManager.class);
dataSourcesSnapshot = EasyMock.createNiceMock(DataSourcesSnapshot.class);
metadataRuleManager = EasyMock.createNiceMock(MetadataRuleManager.class);
@@ -147,11 +137,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
)
).andReturn(new
AtomicReference<>(DruidCompactionConfig.empty())).anyTimes();
EasyMock.replay(configManager);
- setupServerAndCurator();
- curator.start();
- curator.blockUntilConnected();
- curator.create().creatingParentsIfNeeded().forPath(LOADPATH);
- objectMapper = new DefaultObjectMapper();
+ final ObjectMapper objectMapper = new DefaultObjectMapper();
statusTracker = new CompactionStatusTracker(objectMapper);
druidCoordinatorConfig = new DruidCoordinatorConfig(
new CoordinatorRunConfig(new Duration(COORDINATOR_START_DELAY), new
Duration(COORDINATOR_PERIOD)),
@@ -160,26 +146,11 @@ public class DruidCoordinatorTest extends CuratorTestBase
new CostBalancerStrategyFactory(),
null
);
- pathChildrenCache = new PathChildrenCache(
- curator,
- LOADPATH,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache-%d")
- );
- loadQueuePeon = new CuratorLoadQueuePeon(
- curator,
- LOADPATH,
- objectMapper,
-
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_scheduled-%d"),
- Execs.singleThreaded("coordinator_test_load_queue_peon-%d"),
- LOAD_TIMEOUT
- );
- loadQueuePeon.start();
druidNode = new DruidNode("hey", "what", false, 1234, null, true, false);
scheduledExecutorFactory = ScheduledExecutors::fixed;
leaderAnnouncerLatch = new CountDownLatch(1);
leaderUnannouncerLatch = new CountDownLatch(1);
+ serviceEmitter = new LatchableServiceEmitter();
coordinator = new DruidCoordinator(
druidCoordinatorConfig,
createMetadataManager(configManager),
@@ -213,14 +184,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
);
}
- @After
- public void tearDown() throws Exception
- {
- loadQueuePeon.stop();
- pathChildrenCache.close();
- tearDownServerAndCurator();
- }
-
@Test(timeout = 60_000L)
public void testCoordinatorRun() throws Exception
{
@@ -261,7 +224,8 @@ public class DruidCoordinatorTest extends CuratorTestBase
EasyMock.replay(immutableDruidDataSource);
// Setup ServerInventoryView
- druidServer = new DruidServer("server1", "localhost", null, 5L,
ServerType.HISTORICAL, tier, 0);
+ final DruidServer druidServer = new DruidServer("server1", "localhost",
null, 5L, ServerType.HISTORICAL, tier, 0);
+ final LoadQueuePeon loadQueuePeon =
createImmediateLoadPeonFor(druidServer);
setupPeons(Collections.singletonMap("server1", loadQueuePeon));
EasyMock.expect(serverInventoryView.getInventory()).andReturn(
ImmutableList.of(druidServer)
@@ -280,19 +244,8 @@ public class DruidCoordinatorTest extends CuratorTestBase
// This coordinator should be leader by now
Assert.assertTrue(coordinator.isLeader());
Assert.assertEquals(druidNode.getHostAndPort(),
coordinator.getCurrentLeader());
- pathChildrenCache.start();
- final CountDownLatch assignSegmentLatch =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(
- 1,
- pathChildrenCache,
- ImmutableMap.of("2010-01-01T00:00:00.000Z_2010-01-02T00:00:00.000Z",
dataSegment),
- druidServer
- );
- assignSegmentLatch.await();
-
- final CountDownLatch coordinatorRunLatch = new CountDownLatch(2);
- serviceEmitter.latch = coordinatorRunLatch;
- coordinatorRunLatch.await();
+ serviceEmitter.coordinatorRunLatch.await();
Assert.assertEquals(ImmutableMap.of(dataSource, 100.0),
coordinator.getDatasourceToLoadStatus());
@@ -350,42 +303,24 @@ public class DruidCoordinatorTest extends CuratorTestBase
final String dataSource = "dataSource", hotTierName = "hot", coldTierName
= "cold";
final Rule hotTier = new IntervalLoadRule(Intervals.of("2018-01-01/P1M"),
ImmutableMap.of(hotTierName, 1), null);
final Rule coldTier = new ForeverLoadRule(ImmutableMap.of(coldTierName,
1), null);
- final String loadPathCold = "/druid/loadqueue/cold:1234";
final DruidServer hotServer = new DruidServer("hot", "hot", null, 5L,
ServerType.HISTORICAL, hotTierName, 0);
final DruidServer coldServer = new DruidServer("cold", "cold", null, 5L,
ServerType.HISTORICAL, coldTierName, 0);
- final Map<String, DataSegment> dataSegments = ImmutableMap.of(
- "2018-01-02T00:00:00.000Z_2018-01-03T00:00:00.000Z",
+ final Set<DataSegment> dataSegments = Set.of(
new DataSegment(dataSource, Intervals.of("2018-01-02/P1D"), "v1",
null, null, null, null, 0x9, 0),
- "2018-01-03T00:00:00.000Z_2018-01-04T00:00:00.000Z",
new DataSegment(dataSource, Intervals.of("2018-01-03/P1D"), "v1",
null, null, null, null, 0x9, 0),
- "2017-01-01T00:00:00.000Z_2017-01-02T00:00:00.000Z",
new DataSegment(dataSource, Intervals.of("2017-01-01/P1D"), "v1",
null, null, null, null, 0x9, 0)
);
- final LoadQueuePeon loadQueuePeonCold = new CuratorLoadQueuePeon(
- curator,
- loadPathCold,
- objectMapper,
-
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_cold_scheduled-%d"),
- Execs.singleThreaded("coordinator_test_load_queue_peon_cold-%d"),
- LOAD_TIMEOUT
- );
- final PathChildrenCache pathChildrenCacheCold = new PathChildrenCache(
- curator,
- loadPathCold,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache_cold-%d")
- );
- setupPeons(ImmutableMap.of("hot", loadQueuePeon, "cold",
loadQueuePeonCold));
+ final LoadQueuePeon loadQueuePeonHot =
createImmediateLoadPeonFor(hotServer);
+ final LoadQueuePeon loadQueuePeonCold =
createImmediateLoadPeonFor(coldServer);
+ setupPeons(ImmutableMap.of("hot", loadQueuePeonHot, "cold",
loadQueuePeonCold));
+ loadQueuePeonHot.start();
loadQueuePeonCold.start();
- pathChildrenCache.start();
- pathChildrenCacheCold.start();
DruidDataSource[] druidDataSources = {new DruidDataSource(dataSource,
Collections.emptyMap())};
- dataSegments.values().forEach(druidDataSources[0]::addSegment);
+ dataSegments.forEach(druidDataSources[0]::addSegment);
setupSegmentsMetadataMock(druidDataSources[0]);
@@ -402,14 +337,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
coordinator.start();
leaderAnnouncerLatch.await(); // Wait for this coordinator to become leader
- final CountDownLatch assignSegmentLatchHot =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(2,
pathChildrenCache, dataSegments, hotServer);
- final CountDownLatch assignSegmentLatchCold =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(1,
pathChildrenCacheCold, dataSegments, coldServer);
- assignSegmentLatchHot.await();
- assignSegmentLatchCold.await();
-
- final CountDownLatch coordinatorRunLatch = new CountDownLatch(2);
- serviceEmitter.latch = coordinatorRunLatch;
- coordinatorRunLatch.await();
+ serviceEmitter.coordinatorRunLatch.await();
Assert.assertEquals(ImmutableMap.of(dataSource, 100.0),
coordinator.getDatasourceToLoadStatus());
@@ -425,7 +353,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
Assert.assertEquals(0L,
underReplicationCountsPerDataSourcePerTierUsingClusterView.get(hotTierName).getLong(dataSource));
Assert.assertEquals(0L,
underReplicationCountsPerDataSourcePerTierUsingClusterView.get(coldTierName).getLong(dataSource));
- dataSegments.values().forEach(dataSegment ->
Assert.assertEquals(Integer.valueOf(1),
coordinator.getReplicationFactor(dataSegment.getId())));
+ dataSegments.forEach(dataSegment ->
Assert.assertEquals(Integer.valueOf(1),
coordinator.getReplicationFactor(dataSegment.getId())));
coordinator.stop();
leaderUnannouncerLatch.await();
@@ -443,109 +371,39 @@ public class DruidCoordinatorTest extends CuratorTestBase
final String coldTierName = "cold";
final String tierName1 = "tier1";
final String tierName2 = "tier2";
- final String loadPathCold = "/druid/loadqueue/cold:1234";
- final String loadPathBroker1 = "/druid/loadqueue/broker1:1234";
- final String loadPathBroker2 = "/druid/loadqueue/broker2:1234";
- final String loadPathPeon = "/druid/loadqueue/peon:1234";
final DruidServer hotServer = new DruidServer("hot", "hot", null, 5L,
ServerType.HISTORICAL, hotTierName, 0);
final DruidServer coldServer = new DruidServer("cold", "cold", null, 5L,
ServerType.HISTORICAL, coldTierName, 0);
final DruidServer brokerServer1 = new DruidServer("broker1", "broker1",
null, 5L, ServerType.BROKER, tierName1, 0);
final DruidServer brokerServer2 = new DruidServer("broker2", "broker2",
null, 5L, ServerType.BROKER, tierName2, 0);
final DruidServer peonServer = new DruidServer("peon", "peon", null, 5L,
ServerType.INDEXER_EXECUTOR, tierName2, 0);
- final Map<String, DataSegment> dataSegments = ImmutableMap.of(
- "2018-01-02T00:00:00.000Z_2018-01-03T00:00:00.000Z",
+ final Set<DataSegment> dataSegments = Set.of(
new DataSegment(dataSource, Intervals.of("2018-01-02/P1D"), "v1",
null, null, null, null, 0x9, 0),
- "2018-01-03T00:00:00.000Z_2018-01-04T00:00:00.000Z",
new DataSegment(dataSource, Intervals.of("2018-01-03/P1D"), "v1",
null, null, null, null, 0x9, 0),
- "2017-01-01T00:00:00.000Z_2017-01-02T00:00:00.000Z",
new DataSegment(dataSource, Intervals.of("2017-01-01/P1D"), "v1",
null, null, null, null, 0x9, 0)
);
- final LoadQueuePeon loadQueuePeonCold = new CuratorLoadQueuePeon(
- curator,
- loadPathCold,
- objectMapper,
-
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_cold_scheduled-%d"),
- Execs.singleThreaded("coordinator_test_load_queue_peon_cold-%d"),
- LOAD_TIMEOUT
- );
-
- final LoadQueuePeon loadQueuePeonBroker1 = new CuratorLoadQueuePeon(
- curator,
- loadPathBroker1,
- objectMapper,
-
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_broker1_scheduled-%d"),
- Execs.singleThreaded("coordinator_test_load_queue_peon_broker1-%d"),
- LOAD_TIMEOUT
- );
-
- final LoadQueuePeon loadQueuePeonBroker2 = new CuratorLoadQueuePeon(
- curator,
- loadPathBroker2,
- objectMapper,
-
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_broker2_scheduled-%d"),
- Execs.singleThreaded("coordinator_test_load_queue_peon_broker2-%d"),
- LOAD_TIMEOUT
- );
-
- final LoadQueuePeon loadQueuePeonPoenServer = new CuratorLoadQueuePeon(
- curator,
- loadPathPeon,
- objectMapper,
-
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_peon_scheduled-%d"),
- Execs.singleThreaded("coordinator_test_load_queue_peon_peon-%d"),
- LOAD_TIMEOUT
- );
- final PathChildrenCache pathChildrenCacheCold = new PathChildrenCache(
- curator,
- loadPathCold,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache_cold-%d")
- );
- final PathChildrenCache pathChildrenCacheBroker1 = new PathChildrenCache(
- curator,
- loadPathBroker1,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache_broker1-%d")
- );
- final PathChildrenCache pathChildrenCacheBroker2 = new PathChildrenCache(
- curator,
- loadPathBroker2,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache_broker2-%d")
- );
- final PathChildrenCache pathChildrenCachePeon = new PathChildrenCache(
- curator,
- loadPathPeon,
- true,
- true,
- Execs.singleThreaded("coordinator_test_path_children_cache_peon-%d")
- );
-
+ final LoadQueuePeon loadQueuePeonHot =
createImmediateLoadPeonFor(hotServer);
+ final LoadQueuePeon loadQueuePeonCold =
createImmediateLoadPeonFor(coldServer);
+ final LoadQueuePeon loadQueuePeonBroker1 =
createImmediateLoadPeonFor(brokerServer1);
+ final LoadQueuePeon loadQueuePeonBroker2 =
createImmediateLoadPeonFor(brokerServer2);
+ final LoadQueuePeon loadQueuePeonPoenServer =
createImmediateLoadPeonFor(peonServer);
setupPeons(ImmutableMap.of(
- "hot", loadQueuePeon,
+ "hot", loadQueuePeonHot,
"cold", loadQueuePeonCold,
"broker1", loadQueuePeonBroker1,
"broker2", loadQueuePeonBroker2,
"peon", loadQueuePeonPoenServer
));
+ loadQueuePeonHot.start();
loadQueuePeonCold.start();
loadQueuePeonBroker1.start();
loadQueuePeonBroker2.start();
loadQueuePeonPoenServer.start();
- pathChildrenCache.start();
- pathChildrenCacheCold.start();
- pathChildrenCacheBroker1.start();
- pathChildrenCacheBroker2.start();
- pathChildrenCachePeon.start();
DruidDataSource druidDataSource = new DruidDataSource(dataSource,
Collections.emptyMap());
- dataSegments.values().forEach(druidDataSource::addSegment);
+ dataSegments.forEach(druidDataSource::addSegment);
setupSegmentsMetadataMock(druidDataSource);
@@ -563,23 +421,10 @@ public class DruidCoordinatorTest extends CuratorTestBase
coordinator.start();
leaderAnnouncerLatch.await(); // Wait for this coordinator to become leader
- final CountDownLatch assignSegmentLatchHot =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(3,
pathChildrenCache, dataSegments, hotServer);
- final CountDownLatch assignSegmentLatchCold =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(3,
pathChildrenCacheCold, dataSegments, coldServer);
- final CountDownLatch assignSegmentLatchBroker1 =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(3,
pathChildrenCacheBroker1, dataSegments, brokerServer1);
- final CountDownLatch assignSegmentLatchBroker2 =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(3,
pathChildrenCacheBroker2, dataSegments, brokerServer2);
- final CountDownLatch assignSegmentLatchPeon =
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(3,
pathChildrenCachePeon, dataSegments, peonServer);
- assignSegmentLatchHot.await();
- assignSegmentLatchCold.await();
- assignSegmentLatchBroker1.await();
- assignSegmentLatchBroker2.await();
- assignSegmentLatchPeon.await();
-
- final CountDownLatch coordinatorRunLatch = new CountDownLatch(2);
- serviceEmitter.latch = coordinatorRunLatch;
- coordinatorRunLatch.await();
+ serviceEmitter.coordinatorRunLatch.await();
Assert.assertEquals(ImmutableMap.of(dataSource, 100.0),
coordinator.getDatasourceToLoadStatus());
- Assert.assertEquals(new HashSet<>(dataSegments.values()),
coordinator.getBroadcastSegments());
+ Assert.assertEquals(dataSegments, coordinator.getBroadcastSegments());
// Under-replicated counts are updated only after the next coordinator run
Map<String, Object2LongMap<String>>
underReplicationCountsPerDataSourcePerTier =
@@ -875,7 +720,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
// Setup SegmentsMetadataManager
DruidDataSource[] dataSources = {
new DruidDataSource(dataSource, Collections.emptyMap())
-
};
final DataSegment dataSegment = new DataSegment(
dataSource,
@@ -908,11 +752,16 @@ public class DruidCoordinatorTest extends CuratorTestBase
EasyMock.replay(immutableDruidDataSource);
// Setup ServerInventoryView
- druidServer = new DruidServer("server1", "localhost", null, 5L,
ServerType.HISTORICAL, hotTier, 0);
- DruidServer druidServer2 = new DruidServer("server2", "localhost", null,
5L, ServerType.HISTORICAL, coldTier, 0);
- setupPeons(ImmutableMap.of("server1", loadQueuePeon, "server2",
loadQueuePeon));
+ final DruidServer druidServer1 = new DruidServer("server1", "localhost",
null, 5L, ServerType.HISTORICAL, hotTier, 0);
+ final DruidServer druidServer2 = new DruidServer("server2", "localhost",
null, 5L, ServerType.HISTORICAL, coldTier, 0);
+
+ // For hot server, use a load queue peon that does not perform immediate
load
+ setupPeons(ImmutableMap.of(
+ "server1", new TestLoadQueuePeon(),
+ "server2", createImmediateLoadPeonFor(druidServer2))
+ );
EasyMock.expect(serverInventoryView.getInventory()).andReturn(
- ImmutableList.of(druidServer, druidServer2)
+ ImmutableList.of(druidServer1, druidServer2)
).atLeastOnce();
EasyMock.expect(serverInventoryView.isStarted()).andReturn(true).anyTimes();
EasyMock.replay(serverInventoryView, loadQueueTaskMaster);
@@ -925,11 +774,8 @@ public class DruidCoordinatorTest extends CuratorTestBase
// This coordinator should be leader by now
Assert.assertTrue(coordinator.isLeader());
Assert.assertEquals(druidNode.getHostAndPort(),
coordinator.getCurrentLeader());
- pathChildrenCache.start();
- final CountDownLatch coordinatorRunLatch = new CountDownLatch(2);
- serviceEmitter.latch = coordinatorRunLatch;
- coordinatorRunLatch.await();
+ serviceEmitter.coordinatorRunLatch.await();
Object2IntMap<String> numsUnavailableUsedSegmentsPerDataSource =
coordinator.getDatasourceToUnavailableSegmentCount();
@@ -980,33 +826,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
Assert.assertEquals(Collections.emptyMap(), result.getCompactionStates());
}
- private CountDownLatch
createCountDownLatchAndSetPathChildrenCacheListenerWithLatch(
- int latchCount,
- PathChildrenCache pathChildrenCache,
- Map<String, DataSegment> segments,
- DruidServer server
- )
- {
- final CountDownLatch countDownLatch = new CountDownLatch(latchCount);
- pathChildrenCache.getListenable().addListener(
- (CuratorFramework client, PathChildrenCacheEvent event) -> {
- if (CuratorUtils.isChildAdded(event)) {
- DataSegment segment = findSegmentRelatedToCuratorEvent(segments,
event);
- if (segment != null && server.getSegment(segment.getId()) == null)
{
- if (countDownLatch.getCount() > 0) {
- server.addDataSegment(segment);
-
curator.delete().guaranteed().forPath(event.getData().getPath());
- countDownLatch.countDown();
- } else {
- Assert.fail("The segment path " + event.getData().getPath() +
" is not expected");
- }
- }
- }
- }
- );
- return countDownLatch;
- }
-
private void setupSegmentsMetadataMock(DruidDataSource dataSource)
{
EasyMock.expect(segmentsMetadataManager.isPollingDatabasePeriodically()).andReturn(true).anyTimes();
@@ -1041,21 +860,6 @@ public class DruidCoordinatorTest extends CuratorTestBase
EasyMock.replay(this.dataSourcesSnapshot);
}
- @Nullable
- private static DataSegment findSegmentRelatedToCuratorEvent(
- Map<String, DataSegment> dataSegments,
- PathChildrenCacheEvent event
- )
- {
- return dataSegments
- .entrySet()
- .stream()
- .filter(x -> event.getData().getPath().contains(x.getKey()))
- .map(Map.Entry::getValue)
- .findFirst()
- .orElse(null);
- }
-
private void setupPeons(Map<String, LoadQueuePeon> peonMap)
{
loadQueueTaskMaster.resetPeonsForNewServers(EasyMock.anyObject());
@@ -1071,6 +875,18 @@ public class DruidCoordinatorTest extends CuratorTestBase
() -> peonMap.get(((ImmutableDruidServer)
EasyMock.getCurrentArgument(0)).getName())
).anyTimes();
}
+
+ private LoadQueuePeon createImmediateLoadPeonFor(DruidServer server)
+ {
+ return new TestLoadQueuePeon() {
+ @Override
+ public void loadSegment(DataSegment segment, SegmentAction action,
@Nullable LoadPeonCallback callback)
+ {
+ server.addDataSegment(segment);
+ super.loadSegment(segment, action, callback);
+ }
+ };
+ }
private static class TestDruidLeaderSelector implements DruidLeaderSelector
{
@@ -1113,7 +929,7 @@ public class DruidCoordinatorTest extends CuratorTestBase
private static class LatchableServiceEmitter extends ServiceEmitter
{
- private CountDownLatch latch;
+ private final CountDownLatch coordinatorRunLatch = new CountDownLatch(2);
private LatchableServiceEmitter()
{
@@ -1123,8 +939,15 @@ public class DruidCoordinatorTest extends CuratorTestBase
@Override
public void emit(Event event)
{
- if (latch != null &&
"segment/count".equals(event.toMap().get("metric"))) {
- latch.countDown();
+ if (event instanceof ServiceMetricEvent) {
+ final ServiceMetricEvent metricEvent = (ServiceMetricEvent) event;
+
+ // Count down when the historical management duties group has finished
+ String dutyGroupName = (String)
metricEvent.getUserDims().get("dutyGroup");
+ if
(Stats.CoordinatorRun.GROUP_RUN_TIME.getMetricName().equals(metricEvent.getMetric())
+ && "HistoricalManagementDuties".equals(dutyGroupName)) {
+ coordinatorRunLatch.countDown();
+ }
}
}
}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeonTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeonTest.java
deleted file mode 100644
index 4de59af1f2d..00000000000
---
a/server/src/test/java/org/apache/druid/server/coordinator/loading/CuratorLoadQueuePeonTest.java
+++ /dev/null
@@ -1,401 +0,0 @@
-/*
- * 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.server.coordinator.loading;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.base.Function;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Lists;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.recipes.cache.PathChildrenCache;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
-import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
-import org.apache.curator.utils.ZKPaths;
-import org.apache.druid.curator.CuratorTestBase;
-import org.apache.druid.java.util.common.Intervals;
-import org.apache.druid.java.util.common.concurrent.Execs;
-import org.apache.druid.segment.TestHelper;
-import org.apache.druid.server.coordination.DataSegmentChangeCallback;
-import org.apache.druid.server.coordination.DataSegmentChangeHandler;
-import org.apache.druid.server.coordination.DataSegmentChangeRequest;
-import org.apache.druid.server.coordination.SegmentChangeRequestDrop;
-import org.apache.druid.server.coordination.SegmentChangeRequestLoad;
-import org.apache.druid.timeline.DataSegment;
-import org.apache.druid.timeline.SegmentId;
-import org.apache.druid.timeline.partition.NoneShardSpec;
-import org.joda.time.Duration;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-public class CuratorLoadQueuePeonTest extends CuratorTestBase
-{
- private static final String LOAD_QUEUE_PATH =
"/druid/loadqueue/localhost:1234";
-
- private final ObjectMapper jsonMapper = TestHelper.makeJsonMapper();
-
- private CuratorLoadQueuePeon loadQueuePeon;
- private PathChildrenCache loadQueueCache;
-
- @Before
- public void setUp() throws Exception
- {
- setupServerAndCurator();
- curator.start();
- curator.blockUntilConnected();
- curator.create().creatingParentsIfNeeded().forPath(LOAD_QUEUE_PATH);
-
- loadQueueCache = new PathChildrenCache(
- curator,
- LOAD_QUEUE_PATH,
- true,
- true,
- Execs.singleThreaded("load_queue_cache-%d")
- );
- }
-
- @Test
- public void testMultipleLoadDropSegments() throws Exception
- {
- loadQueuePeon = new CuratorLoadQueuePeon(
- curator,
- LOAD_QUEUE_PATH,
- jsonMapper,
- Execs.scheduledSingleThreaded("test_load_queue_peon_scheduled-%d"),
- Execs.singleThreaded("test_load_queue_peon-%d"),
- Duration.standardMinutes(15)
- );
-
- loadQueuePeon.start();
-
- ConcurrentMap<SegmentId, CountDownLatch> loadRequestSignals = new
ConcurrentHashMap<>(5);
- ConcurrentMap<SegmentId, CountDownLatch> dropRequestSignals = new
ConcurrentHashMap<>(5);
- ConcurrentMap<SegmentId, CountDownLatch> segmentLoadedSignals = new
ConcurrentHashMap<>(5);
- ConcurrentMap<SegmentId, CountDownLatch> segmentDroppedSignals = new
ConcurrentHashMap<>(5);
-
- final List<DataSegment> segmentToDrop = Lists.transform(
- ImmutableList.of(
- "2014-10-26T00:00:00Z/P1D",
- "2014-10-25T00:00:00Z/P1D",
- "2014-10-24T00:00:00Z/P1D",
- "2014-10-23T00:00:00Z/P1D",
- "2014-10-22T00:00:00Z/P1D"
- ), new Function<>()
- {
- @Override
- public DataSegment apply(String intervalStr)
- {
- DataSegment dataSegment = dataSegmentWithInterval(intervalStr);
- return dataSegment;
- }
- }
- );
-
- final CountDownLatch[] dropRequestLatches = new CountDownLatch[5];
- final CountDownLatch[] dropSegmentLatches = new CountDownLatch[5];
- for (int i = 0; i < 5; i++) {
- dropRequestLatches[i] = new CountDownLatch(1);
- dropSegmentLatches[i] = new CountDownLatch(1);
- }
- int i = 0;
- for (DataSegment s : segmentToDrop) {
- dropRequestSignals.put(s.getId(), dropRequestLatches[i]);
- segmentDroppedSignals.put(s.getId(), dropSegmentLatches[i++]);
- }
-
- final List<DataSegment> segmentToLoad = Lists.transform(
- ImmutableList.of(
- "2014-10-27T00:00:00Z/P1D",
- "2014-10-29T00:00:00Z/P1M",
- "2014-10-31T00:00:00Z/P1D",
- "2014-10-30T00:00:00Z/P1D",
- "2014-10-28T00:00:00Z/P1D"
- ), new Function<>()
- {
- @Override
- public DataSegment apply(String intervalStr)
- {
- DataSegment dataSegment = dataSegmentWithInterval(intervalStr);
- loadRequestSignals.put(dataSegment.getId(), new CountDownLatch(1));
- segmentLoadedSignals.put(dataSegment.getId(), new
CountDownLatch(1));
- return dataSegment;
- }
- }
- );
-
- final CountDownLatch[] loadRequestLatches = new CountDownLatch[5];
- final CountDownLatch[] segmentLoadedLatches = new CountDownLatch[5];
- for (i = 0; i < 5; i++) {
- loadRequestLatches[i] = new CountDownLatch(1);
- segmentLoadedLatches[i] = new CountDownLatch(1);
- }
- i = 0;
- for (DataSegment s : segmentToDrop) {
- loadRequestSignals.put(s.getId(), loadRequestLatches[i]);
- segmentLoadedSignals.put(s.getId(), segmentLoadedLatches[i++]);
- }
-
- // segment with latest interval should be loaded first
- final List<DataSegment> expectedLoadOrder = Lists.transform(
- ImmutableList.of(
- "2014-10-29T00:00:00Z/P1M",
- "2014-10-31T00:00:00Z/P1D",
- "2014-10-30T00:00:00Z/P1D",
- "2014-10-28T00:00:00Z/P1D",
- "2014-10-27T00:00:00Z/P1D"
- ), intervalStr -> dataSegmentWithInterval(intervalStr)
- );
-
- final DataSegmentChangeHandler handler = new DataSegmentChangeHandler()
- {
- @Override
- public void addSegment(DataSegment segment, DataSegmentChangeCallback
callback)
- {
- loadRequestSignals.get(segment.getId()).countDown();
- }
-
- @Override
- public void removeSegment(DataSegment segment, DataSegmentChangeCallback
callback)
- {
- dropRequestSignals.get(segment.getId()).countDown();
- }
- };
-
- loadQueueCache.getListenable().addListener(
- (client, event) -> {
- if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
- DataSegmentChangeRequest request = jsonMapper.readValue(
- event.getData().getData(),
- DataSegmentChangeRequest.class
- );
- request.go(handler, null);
- }
- }
- );
- loadQueueCache.start();
-
- for (final DataSegment segment : segmentToDrop) {
- loadQueuePeon.dropSegment(
- segment,
- success -> segmentDroppedSignals.get(segment.getId()).countDown()
- );
- }
-
- for (final DataSegment segment : segmentToLoad) {
- loadQueuePeon.loadSegment(
- segment,
- SegmentAction.LOAD,
- success -> segmentLoadedSignals.get(segment.getId()).countDown()
- );
- }
-
- Assert.assertEquals(6000, loadQueuePeon.getSizeOfSegmentsToLoad());
- Assert.assertEquals(5, loadQueuePeon.getSegmentsToLoad().size());
- Assert.assertEquals(5, loadQueuePeon.getSegmentsToDrop().size());
- Assert.assertEquals(0, loadQueuePeon.getTimedOutSegments().size());
-
- for (DataSegment segment : segmentToDrop) {
- String dropRequestPath = ZKPaths.makePath(LOAD_QUEUE_PATH,
segment.getId().toString());
- Assert.assertTrue(
- "Latch not counted down for " +
dropRequestSignals.get(segment.getId()),
- dropRequestSignals.get(segment.getId()).await(10, TimeUnit.SECONDS)
- );
- Assert.assertNotNull(
- "Path " + dropRequestPath + " doesn't exist",
- curator.checkExists().forPath(dropRequestPath)
- );
- Assert.assertEquals(
- segment,
- ((SegmentChangeRequestDrop) jsonMapper.readValue(
- curator.getData()
- .decompressed()
- .forPath(dropRequestPath), DataSegmentChangeRequest.class
- )).getSegment()
- );
-
- // simulate completion of drop request by historical
- curator.delete().guaranteed().forPath(dropRequestPath);
-
Assert.assertTrue(timing.forWaiting().awaitLatch(segmentDroppedSignals.get(segment.getId())));
- }
-
- for (DataSegment segment : expectedLoadOrder) {
- String loadRequestPath = ZKPaths.makePath(LOAD_QUEUE_PATH,
segment.getId().toString());
-
Assert.assertTrue(timing.forWaiting().awaitLatch(loadRequestSignals.get(segment.getId())));
- Assert.assertNotNull(curator.checkExists().forPath(loadRequestPath));
- Assert.assertEquals(
- segment,
- ((SegmentChangeRequestLoad) jsonMapper
-
.readValue(curator.getData().decompressed().forPath(loadRequestPath),
DataSegmentChangeRequest.class))
- .getSegment()
- );
-
- // simulate completion of load request by historical
- curator.delete().guaranteed().forPath(loadRequestPath);
-
Assert.assertTrue(timing.forWaiting().awaitLatch(segmentLoadedSignals.get(segment.getId())));
- }
- }
-
- @Test
- public void testFailAssignForNonTimeoutFailures() throws Exception
- {
- final DataSegment segment =
dataSegmentWithInterval("2014-10-22T00:00:00Z/P1D");
-
- final CountDownLatch segmentLoadedSignal = new CountDownLatch(1);
-
- loadQueuePeon = new CuratorLoadQueuePeon(
- curator,
- LOAD_QUEUE_PATH,
- // This will fail inside SegmentChangeProcessor.run()
- null,
- Execs.scheduledSingleThreaded("test_load_queue_peon_scheduled-%d"),
- Execs.singleThreaded("test_load_queue_peon-%d"),
- // set time-out to 1 ms so that LoadQueuePeon will fail the assignment
quickly
- new Duration(1)
- );
-
- loadQueuePeon.start();
-
- loadQueueCache.start();
-
- loadQueuePeon.loadSegment(
- segment,
- SegmentAction.LOAD,
- success -> segmentLoadedSignal.countDown()
- );
-
- Assert.assertTrue(timing.forWaiting().awaitLatch(segmentLoadedSignal));
- Assert.assertEquals(0, loadQueuePeon.getSegmentsToLoad().size());
- Assert.assertEquals(0L, loadQueuePeon.getSizeOfSegmentsToLoad());
- Assert.assertEquals(0, loadQueuePeon.getTimedOutSegments().size());
-
- }
-
- @Test
- public void testFailAssignForLoadDropTimeout() throws Exception
- {
- final DataSegment segment =
dataSegmentWithInterval("2014-10-22T00:00:00Z/P1D");
-
- final CountDownLatch loadRequestSignal = new CountDownLatch(1);
- final CountDownLatch segmentLoadedSignal = new CountDownLatch(1);
- final CountDownLatch delayedSegmentLoadedSignal = new CountDownLatch(2);
- final CountDownLatch loadRequestRemoveSignal = new CountDownLatch(1);
-
- loadQueuePeon = new CuratorLoadQueuePeon(
- curator,
- LOAD_QUEUE_PATH,
- jsonMapper,
- Execs.scheduledSingleThreaded("test_load_queue_peon_scheduled-%d"),
- Execs.singleThreaded("test_load_queue_peon-%d"),
- // The timeout here was set to 1ms, when this test was acting flakey.
A cursory glance makes me wonder if
- // there's a race where the timeout actually happens before other code
can run. 1ms timeout seems aggressive.
- // 100ms is a great price to pay if it removes the flakeyness,
- new Duration(100)
- );
-
- loadQueuePeon.start();
-
- loadQueueCache.getListenable().addListener(
- new PathChildrenCacheListener()
- {
- @Override
- public void childEvent(CuratorFramework client,
PathChildrenCacheEvent event)
- {
- switch (event.getType()) {
- case CHILD_ADDED:
- loadRequestSignal.countDown();
- break;
- case CHILD_REMOVED:
- loadRequestRemoveSignal.countDown();
- break;
- default:
- // pass
- }
- }
- }
- );
- loadQueueCache.start();
-
- loadQueuePeon.loadSegment(
- segment,
- SegmentAction.LOAD,
- success -> {
- segmentLoadedSignal.countDown();
- delayedSegmentLoadedSignal.countDown();
- }
- );
-
- String loadRequestPath = ZKPaths.makePath(LOAD_QUEUE_PATH,
segment.getId().toString());
-
- Assert.assertTrue(timing.forWaiting().awaitLatch(loadRequestSignal));
- Assert.assertNotNull(curator.checkExists().forPath(loadRequestPath));
- Assert.assertEquals(
- segment,
- ((SegmentChangeRequestLoad) jsonMapper
-
.readValue(curator.getData().decompressed().forPath(loadRequestPath),
DataSegmentChangeRequest.class))
- .getSegment()
- );
-
- // simulate incompletion of load request since request has timed out
- Assert.assertTrue(timing.forWaiting().awaitLatch(segmentLoadedSignal));
- Assert.assertEquals(1, loadQueuePeon.getSegmentsToLoad().size());
- Assert.assertEquals(1200L, loadQueuePeon.getSizeOfSegmentsToLoad());
- Assert.assertEquals(1, loadQueuePeon.getTimedOutSegments().size());
-
- // simulate completion of load request by historical after time out on
coordinator
- curator.delete().guaranteed().forPath(loadRequestPath);
-
Assert.assertTrue(timing.forWaiting().awaitLatch(delayedSegmentLoadedSignal));
- Assert.assertTrue(timing.forWaiting().awaitLatch(loadRequestRemoveSignal));
- Assert.assertEquals(0, loadQueuePeon.getSegmentsToLoad().size());
- Assert.assertEquals(0L, loadQueuePeon.getSizeOfSegmentsToLoad());
- Assert.assertEquals(0, loadQueuePeon.getTimedOutSegments().size());
-
- }
-
- private DataSegment dataSegmentWithInterval(String intervalStr)
- {
- return DataSegment.builder()
- .dataSource("test_load_queue_peon")
- .interval(Intervals.of(intervalStr))
- .loadSpec(ImmutableMap.of())
- .version("2015-05-27T03:38:35.683Z")
- .dimensions(ImmutableList.of())
- .metrics(ImmutableList.of())
- .shardSpec(NoneShardSpec.instance())
- .binaryVersion(9)
- .size(1200)
- .build();
- }
-
- @After
- public void tearDown() throws Exception
- {
- loadQueueCache.close();
- loadQueuePeon.stop();
- tearDownServerAndCurator();
- }
-}
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/README.md
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/README.md
index 9e827083284..cd925ccfa72 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/README.md
+++
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/README.md
@@ -74,7 +74,6 @@ of the coordinator in these situations.
- It should not be used to verify the absolute values of execution latencies,
e.g. the time taken to compute the
balancing cost of a segment. But the relative values can still be a good
indicator while doing comparisons between,
say two balancing strategies.
-- It does not support simulation of the zk-based `CuratorLoadQueuePeon`.
## Usage
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
