Skip to content

fix issue with CuratorLoadQueuePeon shutting down executors it does not own #8140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jon-wei merged 3 commits into from
Jul 24, 2019
Merged

fix issue with CuratorLoadQueuePeon shutting down executors it does not own #8140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d5cb8e1
fix issue with CuratorLoadQueuePeon shutting down executors it does n…
clintropolis Jul 23, 2019
440f250
use lifecycled executors
clintropolis Jul 23, 2019
34096a2
maybe this
clintropolis Jul 23, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions server/src/main/java/org/apache/druid/server/coordinator/CuratorLoadQueuePeon.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -335,8 +335,6 @@ public void stop()

queuedSize.set(0L);
failedAssignCount.set(0);
processingExecutor.shutdown();
Copy link
Copy Markdown
Contributor

@jon-wei jon-wei Jul 23, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we shut these down somewhere? (Maybe tie LoadQueueTaskMaster to lifecycle and shutdown the execs there?)

Copy link
Copy Markdown
Member Author

@clintropolis clintropolis Jul 23, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That seems reasonable, 👍

Copy link
Copy Markdown
Member Author

@clintropolis clintropolis Jul 23, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Used the factory passed into the method in CliCoordinator to create the executors, which itself appears to be under lifecycle.

Copy link
Copy Markdown
Member Author

@clintropolis clintropolis Jul 23, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, that won't work...

Copy link
Copy Markdown
Member Author

@clintropolis clintropolis Jul 24, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, that won't work...

Fixed to use ExecutorServices.manageLifecycle to tie the executors to the main service lifecycle.

callBackExecutor.shutdown();
}

private void entryRemoved(SegmentHolder segmentHolder, String path)
Expand Down
22 changes: 18 additions & 4 deletions server/src/test/java/org/apache/druid/server/coordinator/CuratorDruidCoordinatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,9 @@
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

Expand Down Expand Up @@ -119,6 +121,9 @@ public class CuratorDruidCoordinatorTest extends CuratorTestBase
private final ObjectMapper jsonMapper;
private final ZkPathsConfig zkPathsConfig;

private ScheduledExecutorService peonExec = Execs.scheduledSingleThreaded("Master-PeonExec--%d");
private ExecutorService callbackExec = Execs.multiThreaded(4, "LoadQueuePeon-callbackexec--%d");

public CuratorDruidCoordinatorTest()
{
jsonMapper = TestHelper.makeJsonMapper();
Expand Down Expand Up @@ -187,16 +192,16 @@ public void setUp() throws Exception
curator,
SOURCE_LOAD_PATH,
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_src_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon_src-%d"),
peonExec,
callbackExec,
druidCoordinatorConfig
);
destinationLoadQueuePeon = new CuratorLoadQueuePeon(
curator,
DESTINATION_LOAD_PATH,
objectMapper,
Execs.scheduledSingleThreaded("coordinator_test_load_queue_peon_dest_scheduled-%d"),
Execs.singleThreaded("coordinator_test_load_queue_peon_dest-%d"),
peonExec,
callbackExec,
druidCoordinatorConfig
);
druidNode = new DruidNode("hey", "what", false, 1234, null, true, false);
Expand Down Expand Up @@ -261,6 +266,15 @@ public void tearDown() throws Exception
@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
{
Expand Down
12 changes: 9 additions & 3 deletions services/src/main/java/org/apache/druid/cli/CliCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@
import org.apache.druid.guice.annotations.EscalatedGlobal;
import org.apache.druid.guice.http.JettyHttpClientModule;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.concurrent.ExecutorServices;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutorFactory;
import org.apache.druid.java.util.common.lifecycle.Lifecycle;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.metadata.MetadataRuleManager;
Expand Down Expand Up @@ -248,17 +250,21 @@ public LoadQueueTaskMaster getLoadQueueTaskMaster(
ScheduledExecutorFactory factory,
DruidCoordinatorConfig config,
@EscalatedGlobal HttpClient httpClient,
ZkPathsConfig zkPaths
ZkPathsConfig zkPaths,
Lifecycle lifecycle
)
{
boolean useHttpLoadQueuePeon = "http".equalsIgnoreCase(config.getLoadQueuePeonType());
ExecutorService callBackExec;
if (useHttpLoadQueuePeon) {
callBackExec = Execs.singleThreaded("LoadQueuePeon-callbackexec--%d");
} else {
callBackExec = Execs.multiThreaded(config.getNumCuratorCallBackThreads(), "LoadQueuePeon"
+ "-callbackexec--%d");
callBackExec = Execs.multiThreaded(
config.getNumCuratorCallBackThreads(),
"LoadQueuePeon-callbackexec--%d"
);
}
ExecutorServices.manageLifecycle(lifecycle, callBackExec);
return new LoadQueueTaskMaster(
curator,
jsonMapper,
Expand Down