Skip to content

[improve][meta] Fix invalid use of drain API and race condition in closing metadata store #22585

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
merlimat merged 2 commits into from
Jun 14, 2024
Merged

[improve][meta] Fix invalid use of drain API and race condition in closing metadata store #22585

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
edb72fd
[improve][meta] Fix race condition in closing metadata store
lhotari Apr 25, 2024
a34b44a
Merge remote-tracking branch 'apache/master' into lh-fix-metadata-sto…
merlimat Jun 13, 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
29 changes: 25 additions & 4 deletions .../src/main/java/org/apache/pulsar/metadata/impl/batching/AbstractBatchedMetadataStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,9 +86,13 @@ public void close() throws Exception {
// Fail all the pending items
MetadataStoreException ex =
new MetadataStoreException.AlreadyClosedException("Metadata store is getting closed");
readOps.drain(op -> op.getFuture().completeExceptionally(ex));
writeOps.drain(op -> op.getFuture().completeExceptionally(ex));

MetadataOp op;
while ((op = readOps.poll()) != null) {
op.getFuture().completeExceptionally(ex);
}
while ((op = writeOps.poll()) != null) {
op.getFuture().completeExceptionally(ex);
}
scheduledTask.cancel(true);
}
super.close();
Expand All @@ -98,7 +102,13 @@ public void close() throws Exception {
private void flush() {
while (!readOps.isEmpty()) {
List<MetadataOp> ops = new ArrayList<>();
readOps.drain(ops::add, maxOperations);
for (int i = 0; i < maxOperations; i++) {
MetadataOp op = readOps.poll();
if (op == null) {
break;
}
ops.add(op);
}
internalBatchOperation(ops);
}

Expand Down Expand Up @@ -167,6 +177,11 @@ public void updateMetadataEventSynchronizer(MetadataEventSynchronizer synchroniz
}

private void enqueue(MessagePassingQueue<MetadataOp> queue, MetadataOp op) {
if (isClosed()) {
MetadataStoreException ex = new MetadataStoreException.AlreadyClosedException();
op.getFuture().completeExceptionally(ex);
return;
}
if (enabled) {
if (!queue.offer(op)) {
// Execute individually if we're failing to enqueue
Expand All @@ -182,6 +197,12 @@ private void enqueue(MessagePassingQueue<MetadataOp> queue, MetadataOp op) {
}

private void internalBatchOperation(List<MetadataOp> ops) {
if (isClosed()) {
MetadataStoreException ex =
new MetadataStoreException.AlreadyClosedException();
ops.forEach(op -> op.getFuture().completeExceptionally(ex));
return;
}
long now = System.currentTimeMillis();
for (MetadataOp op : ops) {
this.batchMetadataStoreStats.recordOpWaiting(now - op.created());
Expand Down