Skip to content
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

Jump to bottom

[fix][meta] Use ForkJoinPool.commonPool to handle Metadata operations #22542

Closed
wants to merge 2 commits into from
Closed

[fix][meta] Use ForkJoinPool.commonPool to handle Metadata operations #22542

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
76c1b4e
[fix][broker] Use ForkJoinPool.commonPool to handle Metadata operations
heesung-sn Apr 19, 2024
d832553
added ut
heesung-sn Apr 19, 2024
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
28 changes: 16 additions & 12 deletions pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -542,27 +542,31 @@ public void invalidateCaches(String...paths) {
}

/**
* Run the task in the executor thread and fail the future if the executor is shutting down.
* Run the task in the ForkJoinPool.commonPool thread and fail the future if exceptionally.
*/
@VisibleForTesting
public void execute(Runnable task, CompletableFuture<?> future) {
try {
executor.execute(task);
} catch (Throwable t) {
future.completeExceptionally(t);
}
CompletableFuture.runAsync(task).exceptionally(e -> {
if (!future.isDone()) {
future.completeExceptionally(e);
}
return null;
});
}

/**
* Run the task in the executor thread and fail the future if the executor is shutting down.
* Run the task in the ForkJoinPool.commonPool thread and fail the future if exceptionally.
*/
@VisibleForTesting
public void execute(Runnable task, Supplier<List<CompletableFuture<?>>> futures) {
try {
executor.execute(task);
} catch (final Throwable t) {
futures.get().forEach(f -> f.completeExceptionally(t));
}
CompletableFuture.runAsync(task).exceptionally(e -> {
futures.get().forEach(f -> {
if (!f.isDone()) {
f.completeExceptionally(e);
}
});
return null;
});
}

protected static String parent(String path) {
Expand Down
22 changes: 18 additions & 4 deletions pulsar-metadata/src/test/java/org/apache/pulsar/metadata/MetadataStoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -443,7 +443,7 @@ public Object[][] conditionOfSwitchThread(){
};
}

@Test(dataProvider = "conditionOfSwitchThread")
@Test(dataProvider = "conditionOfSwitchThread", invocationTimeOut = 30_000)
public void testThreadSwitchOfZkMetadataStore(boolean hasSynchronizer, boolean enabledBatch) throws Exception {
final String prefix = newKey();
final String metadataStoreName = UUID.randomUUID().toString().replaceAll("-", "");
Expand All @@ -458,11 +458,12 @@ public void testThreadSwitchOfZkMetadataStore(boolean hasSynchronizer, boolean e
@Cleanup
ZKMetadataStore store = (ZKMetadataStore) MetadataStoreFactory.create(zks.getConnectionString(), config);

String forkJoinPoolNamePrefix = "ForkJoinPool.commonPool";
final Runnable verify = () -> {
String currentThreadName = Thread.currentThread().getName();
String errorMessage = String.format("Expect to switch to thread %s, but currently it is thread %s",
metadataStoreName, currentThreadName);
assertTrue(Thread.currentThread().getName().startsWith(metadataStoreName), errorMessage);
String errorMessage = String.format("Expect to switch to thread %s*, but currently it is thread %s",
forkJoinPoolNamePrefix, currentThreadName);
assertTrue(currentThreadName.startsWith(forkJoinPoolNamePrefix), errorMessage);
};

// put with node which has parent(but the parent node is not exists).
Expand All @@ -480,6 +481,19 @@ public void testThreadSwitchOfZkMetadataStore(boolean hasSynchronizer, boolean e
verify.run();
return null;
}).join();
// blocking chaining call
store.get(prefix + "/d1").thenApply((ignore) -> {
try {
verify.run();
return store.get(prefix + "/e1").thenApply((ignore2) -> {
verify.run();
return null;
}).get();
Copy link
Contributor

Choose a reason for hiding this comment

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

The problem here is a mix between sync & async calls.
eg: blocking in a callback thread.

Using ForkJoinPool, it's only moving the problem to a different place. It will be able to avoid deadlocking on a simple 1:1 situation, though when the ForkJoinPool threads count is maxed, we will still be subject to deadlock.

No matter how we handle this, it's going to be problematic, hence we always have had to be careful in avoiding blocking calls to metadata.

} catch (Exception e) {
throw new RuntimeException(e);
}
}
).get(5, TimeUnit.SECONDS);
// get the node which is not exists.
store.get(prefix + "/non").thenApply((ignore) -> {
verify.run();
Expand Down
Loading