Fixed race condition intruduced managed ledger addEntry introduced in
@@ -95,7 +95,6 @@ import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.InitialPosition;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
-import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -205,7 +204,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
* Queue of pending entries to be added to the managed ledger. Typically entries are queued when a new ledger is
* created asynchronously and hence there is no ready ledger to write into.
*/
- final GrowableArrayBlockingQueue<OpAddEntry> pendingAddEntries = new GrowableArrayBlockingQueue<>();
+ final ConcurrentLinkedQueue<OpAddEntry> pendingAddEntries = new ConcurrentLinkedQueue<>();
// //////////////////////////////////////////////////////////////////////
@@ -491,10 +490,11 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
OpAddEntry addOperation = OpAddEntry.create(this, buffer, callback, ctx);
- pendingAddEntries.add(addOperation);
// Jump to specific thread to avoid contention from writers writing from different threads
executor.executeOrdered(name, safeRun(() -> {
+ pendingAddEntries.add(addOperation);
+
internalAsyncAddEntry(addOperation);
}));
}
@@ -1200,7 +1200,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
// Process all the pending addEntry requests
- for (OpAddEntry op : pendingAddEntries.toList()) {
+ for (OpAddEntry op : pendingAddEntries) {
op.setLedger(currentLedger);
++currentLedgerEntries;
currentLedgerSize += op.data.readableBytes();
