pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [pulsar] branch master updated: Support Add-entry timeout at broker to avoid stuck topics (#3347)
Date Tue, 15 Jan 2019 15:16:53 GMT
This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new a704afd  Support Add-entry timeout at broker to avoid stuck topics (#3347)
a704afd is described below

commit a704afd1946c5648f06b12031320359d5aef623d
Author: Rajan Dhabalia <rdhabalia@apache.org>
AuthorDate: Tue Jan 15 07:16:41 2019 -0800

    Support Add-entry timeout at broker to avoid stuck topics (#3347)
    
    ### Motivation
    
    Recently and in past, we have seen few instances where bookie crashes when it goes out
of memory and kernel panics and broker doesn't complete add-entry callback because of that
topics get stuck on pending-writes.
    
    ```
    2010-01-01T00:18:19+0000 2010-01-01 00:18:19,594 21907 report_stuck:556 WARNING persistent://prop1/global/prod-rt-mxt/extractions-partition-0
is stuck (broker1.us-west1.com): waiting for write
    2010-01-01T00:18:22+0000 2010-01-01 00:18:22,654 21907 report_stuck:556 WARNING persistent://prop1/global/mbr-events/27_bf1
is stuck (broker1.us-west1.com): waiting for write
    2010-01-01T00:19:10+0000 2010-01-01 00:19:10,790 21907 report_stuck:556 WARNING persistent://prop1/global/mbr-events/30_bf1
is stuck (broker1.us-west1.com): waiting for write
    2010-01-01T00:19:14+0000 2010-01-01 00:19:14,692 21907 report_stuck:556 WARNING persistent://prop1/global/mbr-events/35_gq2
is stuck (broker1.us-west1.com): waiting for write
    2010-01-01T00:19:15+0000 2010-01-01 00:19:15,809 21907 report_stuck:556 WARNING persistent://prop1/global/jedi-events/batchevents-partition-35
is stuck (broker1.us-west1.com): waiting for write
    2010-01-01T00:19:25+0000 2010-01-01 00:19:25,127 21907 report_stuck:556 WARNING persistent://prop1/global/jedi-events/mailevents-partition-24
is stuck (broker1.us-west1.com): waiting for write
    ```
    
    internal-stats
    ```
    {
      "entriesAddedCounter" : 66066317,
      "numberOfEntries" : 14129359,
      "totalSize" : 145505085493,
      "currentLedgerEntries" : 238938,
      "currentLedgerSize" : 3715327400,
      "lastLedgerCreatedTimestamp" : "2018-11-27 00:03:03.125+0000",
      "waitingCursorsCount" : 0,
      "pendingAddEntriesCount" : 252514,
      "lastConfirmedEntry" : "489781963:26121",
      "state" : "ClosingLedger",
      "ledgers" : [ {
        "ledgerId" : 489765238,
        "entries" : 559855,
        "size" : 6928710817
      }, {
    ...
    ```
    
    ### Modifications
    
    Add support of write timeout which can be disable by configuring timeout=0.
    
    ### Result
    
    topic will not stuck in pending write in case add-entry callback doesn't completes.
---
 conf/broker.conf                                   |  3 +
 conf/standalone.conf                               |  3 +
 .../bookkeeper/mledger/ManagedLedgerConfig.java    | 16 ++++-
 .../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 30 ++++++++
 .../apache/bookkeeper/mledger/impl/OpAddEntry.java | 67 +++++++++++++++---
 .../bookkeeper/client/PulsarMockLedgerHandle.java  |  2 +-
 .../bookkeeper/mledger/impl/ManagedLedgerTest.java | 82 ++++++++++++++++++++++
 .../apache/pulsar/broker/ServiceConfiguration.java |  9 ++-
 .../pulsar/broker/service/BrokerService.java       |  1 +
 .../broker/auth/MockedPulsarServiceBaseTest.java   |  7 ++
 .../client/impl/BrokerClientIntegrationTest.java   | 76 +++++++++++++++++++-
 site/_data/config/broker.yaml                      |  3 +
 12 files changed, 282 insertions(+), 17 deletions(-)

diff --git a/conf/broker.conf b/conf/broker.conf
index 1cf3a8a..3c2e5ac 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -445,6 +445,9 @@ managedLedgerMetadataOperationsTimeoutSeconds=60
 # Read entries timeout when broker tries to read messages from bookkeeper.
 managedLedgerReadEntryTimeoutSeconds=120
 
+# Add entry timeout when broker tries to publish message to bookkeeper (0 to disable it).
+managedLedgerAddEntryTimeoutSeconds=120
+
 ### --- Load balancer --- ###
 
 # Enable load balancer
diff --git a/conf/standalone.conf b/conf/standalone.conf
index 16d6465..06be447 100644
--- a/conf/standalone.conf
+++ b/conf/standalone.conf
@@ -332,6 +332,9 @@ managedLedgerMetadataOperationsTimeoutSeconds=60
 # Read entries timeout when broker tries to read messages from bookkeeper.
 managedLedgerReadEntryTimeoutSeconds=120
 
+# Add entry timeout when broker tries to publish message to bookkeeper (0 to disable it).
+managedLedgerAddEntryTimeoutSeconds=120
+
 ### --- Load balancer --- ###
 
 loadManagerClassName=org.apache.pulsar.broker.loadbalance.NoopLoadManager
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
index 255b534..8f89050 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerConfig.java
@@ -58,7 +58,7 @@ public class ManagedLedgerConfig {
     private long offloadAutoTriggerSizeThresholdBytes = -1;
     private long metadataOperationsTimeoutSeconds = 60;
     private long readEntryTimeoutSeconds = 120;
-
+    private long addEntryTimeoutSeconds = 120;
     private DigestType digestType = DigestType.CRC32C;
     private byte[] password = "".getBytes(Charsets.UTF_8);
     private LedgerOffloader ledgerOffloader = NullLedgerOffloader.INSTANCE;
@@ -554,4 +554,18 @@ public class ManagedLedgerConfig {
         this.readEntryTimeoutSeconds = readEntryTimeoutSeconds;
         return this;
     }
+    
+    public long getAddEntryTimeoutSeconds() {
+        return addEntryTimeoutSeconds;
+    }
+
+    /**
+     * Add-entry timeout after which add-entry callback will be failed if add-entry is not
succeeded.
+     * 
+     * @param addEntryTimeoutSeconds
+     */
+    public ManagedLedgerConfig setAddEntryTimeoutSeconds(long addEntryTimeoutSeconds) {
+        this.addEntryTimeoutSeconds = addEntryTimeoutSeconds;
+        return this;
+    }
 }
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index dd54630..26b3acc 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -20,6 +20,7 @@ package org.apache.bookkeeper.mledger.impl;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.lang.Math.min;
+import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.FALSE;
 import static org.apache.bookkeeper.mledger.util.SafeRun.safeRun;
 
 import java.time.Clock;
@@ -105,6 +106,7 @@ import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.BoundType;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
@@ -166,6 +168,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback
{
     final Map<String, CompletableFuture<ManagedCursor>> uninitializedCursors;
 
     final EntryCache entryCache;
+    
+    private ScheduledFuture<?> timeoutTask;
 
     /**
      * This lock is held while the ledgers list is updated asynchronously on the metadata
store. Since we use the store
@@ -328,6 +332,28 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback
{
                 }
             }
         });
+        
+        scheduleTimeoutTask();
+    }
+
+    private void scheduleTimeoutTask() {
+        long timeoutSec = config.getAddEntryTimeoutSeconds();
+        // disable timeout task checker if timeout <= 0
+        if (timeoutSec > 0) {
+            this.timeoutTask = this.scheduledExecutor.scheduleAtFixedRate(() -> {
+                OpAddEntry opAddEntry = pendingAddEntries.peek();
+                if (opAddEntry != null) {
+                    boolean isTimedOut = opAddEntry.lastInitTime != -1
+                            && TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() -
opAddEntry.lastInitTime) >= timeoutSec
+                            && opAddEntry.completed == FALSE;
+                    if (isTimedOut) {
+                        log.error("Failed to add entry for ledger {} in time-out {} sec",
+                                (opAddEntry.ledger != null ? opAddEntry.ledger.getId() :
-1), timeoutSec);
+                        opAddEntry.handleAddFailure(opAddEntry.ledger);
+                    }
+                }
+            }, config.getAddEntryTimeoutSeconds(), config.getAddEntryTimeoutSeconds(), TimeUnit.SECONDS);
+        }
     }
 
     private synchronized void initializeBookKeeper(final ManagedLedgerInitializeLedgerCallback
callback) {
@@ -1146,6 +1172,10 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback
{
 
             closeAllCursors(callback, ctx);
         }, null);
+        
+        if (this.timeoutTask != null) {
+            this.timeoutTask.cancel(false);
+        }
     }
 
     private void closeAllCursors(CloseCallback callback, final Object ctx) {
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index 5d9d57c..0cef300 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -23,7 +23,10 @@ import static com.google.common.base.Preconditions.checkArgument;
 import io.netty.buffer.ByteBuf;
 import io.netty.util.Recycler;
 import io.netty.util.Recycler.Handle;
+
+import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
 import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
@@ -35,6 +38,8 @@ import org.apache.bookkeeper.mledger.util.SafeRun;
 import org.apache.bookkeeper.util.SafeRunnable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.TRUE;
+import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.FALSE;
 
 /**
  * Handles the life-cycle of an addEntry() operation.
@@ -42,7 +47,7 @@ import org.slf4j.LoggerFactory;
  */
 class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback {
     private ManagedLedgerImpl ml;
-    private LedgerHandle ledger;
+    LedgerHandle ledger;
     private long entryId;
 
     @SuppressWarnings("unused")
@@ -50,6 +55,11 @@ class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback
{
     private Object ctx;
     private boolean closeWhenDone;
     private long startTime;
+    volatile long lastInitTime;
+    private static final AtomicIntegerFieldUpdater<OpAddEntry> COMPLETED_UPDATER =
+        AtomicIntegerFieldUpdater.newUpdater(OpAddEntry.class, "completed");
+    @SuppressWarnings("unused")
+    volatile int completed = FALSE;
     ByteBuf data;
     private int dataLength;
 
@@ -67,6 +77,7 @@ class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback
{
         op.closeWhenDone = false;
         op.entryId = -1;
         op.startTime = System.nanoTime();
+        op.completed = FALSE;
         ml.mbean.addAddEntrySample(op.dataLength);
         if (log.isDebugEnabled()) {
             log.debug("Created new OpAddEntry {}", op);
@@ -86,6 +97,7 @@ class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback
{
         ByteBuf duplicateBuffer = data.retainedDuplicate();
         // duplicatedBuffer has refCnt=1 at this point
 
+        lastInitTime = System.nanoTime();
         ledger.asyncAddEntry(duplicateBuffer, this, ctx);
 
         // Internally, asyncAddEntry() is refCnt neutral to respect to the passed buffer
and it will keep a ref on it
@@ -95,6 +107,9 @@ class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback
{
     }
 
     public void failed(ManagedLedgerException e) {
+        if (!checkAndCompleteTimeoutTask()) {
+            return;
+        }
         AddEntryCallback cb = callbackUpdater.getAndSet(this, null);
         data.release();
         if (cb != null) {
@@ -119,17 +134,11 @@ class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback
{
         }
 
         if (rc != BKException.Code.OK) {
-            // If we get a write error, we will try to create a new ledger and re-submit
the pending writes. If the
-            // ledger creation fails (persistent bk failure, another instanche owning the
ML, ...), then the writes will
-            // be marked as failed.
-            ml.mbean.recordAddEntryError();
-
-            ml.getExecutor().executeOrdered(ml.getName(), SafeRun.safeRun(() -> {
-                // Force the creation of a new ledger. Doing it in a background thread to
avoid acquiring ML lock
-                // from a BK callback.
-                ml.ledgerClosed(lh);
-            }));
+            handleAddFailure(lh);
         } else {
+            if(!checkAndCompleteTimeoutTask()) {
+                return;
+            }
             // Trigger addComplete callback in a thread hashed on the managed ledger name
             ml.getExecutor().executeOrdered(ml.getName(), this);
         }
@@ -200,6 +209,40 @@ class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback
{
         ml.mbean.addAddEntryLatencySample(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
     }
 
+    /**
+     * It cancels timeout task and checks if add-entry operation is not completed yet.
+     * 
+     * @return true if task is not already completed else returns false.
+     */
+    private boolean checkAndCompleteTimeoutTask() {
+        if (!COMPLETED_UPDATER.compareAndSet(this, FALSE, TRUE)) {
+            if (log.isDebugEnabled()) {
+                log.debug("Add-entry already completed for {}-{}", this.ledger != null ?
this.ledger.getId() : -1,
+                        this.entryId);
+            }
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * It handles add failure on the given ledger. it can be triggered when add-entry fails
or times out.
+     * 
+     * @param ledger
+     */
+    void handleAddFailure(final LedgerHandle ledger) {
+        // If we get a write error, we will try to create a new ledger and re-submit the
pending writes. If the
+        // ledger creation fails (persistent bk failure, another instanche owning the ML,
...), then the writes will
+        // be marked as failed.
+        ml.mbean.recordAddEntryError();
+
+        ml.getExecutor().executeOrdered(ml.getName(), SafeRun.safeRun(() -> {
+            // Force the creation of a new ledger. Doing it in a background thread to avoid
acquiring ML lock
+            // from a BK callback.
+            ml.ledgerClosed(ledger);
+        }));
+    }
+    
     private final Handle<OpAddEntry> recyclerHandle;
 
     private OpAddEntry(Handle<OpAddEntry> recyclerHandle) {
@@ -220,8 +263,10 @@ class OpAddEntry extends SafeRunnable implements AddCallback, CloseCallback
{
         callback = null;
         ctx = null;
         closeWhenDone = false;
+        completed = FALSE;
         entryId = -1;
         startTime = -1;
+        lastInitTime = -1;
         recyclerHandle.recycle(this);
     }
 
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
index cedd3b0..2397a62 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/client/PulsarMockLedgerHandle.java
@@ -59,7 +59,7 @@ public class PulsarMockLedgerHandle extends LedgerHandle {
     long lastEntry = -1;
     boolean fenced = false;
 
-    PulsarMockLedgerHandle(PulsarMockBookKeeper bk, long id,
+    public PulsarMockLedgerHandle(PulsarMockBookKeeper bk, long id,
                            DigestType digest, byte[] passwd) throws GeneralSecurityException
{
         super(bk, id, new LedgerMetadata(3, 3, 2, DigestType.MAC, "".getBytes()), DigestType.MAC,
"".getBytes(),
                 EnumSet.noneOf(WriteFlag.class));
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index a16dcd2..6ed13a5 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -22,6 +22,7 @@ import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyInt;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
@@ -34,8 +35,10 @@ import java.io.IOException;
 import java.lang.reflect.Field;
 import java.lang.reflect.Modifier;
 import java.nio.charset.Charset;
+import java.security.GeneralSecurityException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
@@ -46,10 +49,12 @@ import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
 
+import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
 import org.apache.bookkeeper.client.AsyncCallback.CreateCallback;
 import org.apache.bookkeeper.client.BKException;
 import org.apache.bookkeeper.client.BookKeeper;
@@ -57,6 +62,13 @@ import org.apache.bookkeeper.client.BookKeeper.DigestType;
 import org.apache.bookkeeper.client.LedgerHandle;
 import org.apache.bookkeeper.client.api.LedgerEntries;
 import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.LedgerMetadata;
+import org.apache.bookkeeper.client.PulsarMockBookKeeper;
+import org.apache.bookkeeper.client.PulsarMockLedgerHandle;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.client.api.WriteFlag;
+import org.apache.bookkeeper.conf.ClientConfiguration;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.AddEntryCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteLedgerCallback;
@@ -2324,4 +2336,74 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
 
         ledger.close();
     }
+
+    /**
+     * It verifies that if bk-client doesn't complete the add-entry in given time out then
broker is resilient enought
+     * to create new ledger and add entry successfully.
+     * 
+     * 
+     * @throws Exception
+     */
+    @Test(timeOut = 20000)
+    public void testManagedLedgerWithAddEntryTimeOut() throws Exception {
+        ManagedLedgerConfig config = new ManagedLedgerConfig().setAddEntryTimeoutSeconds(1);
+        ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open("timeout_ledger_test",
config);
+
+        BookKeeper bk = mock(BookKeeper.class);
+        doNothing().when(bk).asyncCreateLedger(anyInt(), anyInt(), anyInt(), any(), any(),
any(), any(), any());
+
+        PulsarMockBookKeeper bkClient = mock(PulsarMockBookKeeper.class);
+        ClientConfiguration conf = new ClientConfiguration();
+        doReturn(conf).when(bkClient).getConf();
+        class MockLedgerHandle extends PulsarMockLedgerHandle {
+            public MockLedgerHandle(PulsarMockBookKeeper bk, long id, DigestType digest,
byte[] passwd)
+                    throws GeneralSecurityException {
+                super(bk, id, digest, passwd);
+            }
+
+            @Override
+            public void asyncAddEntry(final byte[] data, final AddCallback cb, final Object
ctx) {
+                // do nothing
+            }
+
+            @Override
+            public void asyncClose(org.apache.bookkeeper.client.AsyncCallback.CloseCallback
cb, Object ctx) {
+                cb.closeComplete(BKException.Code.OK, this, ctx);
+            }
+        }
+        MockLedgerHandle ledgerHandle = mock(MockLedgerHandle.class);
+        final String data = "data";
+        doNothing().when(ledgerHandle).asyncAddEntry(data.getBytes(), null, null);
+        AtomicBoolean addSuccess = new AtomicBoolean();
+
+        setFieldValue(ManagedLedgerImpl.class, ledger, "currentLedger", ledgerHandle);
+
+        final int totalAddEntries = 1;
+        CountDownLatch latch = new CountDownLatch(totalAddEntries);
+        ledger.asyncAddEntry(data.getBytes(), new AddEntryCallback() {
+
+            @Override
+            public void addComplete(Position position, Object ctx) {
+                addSuccess.set(true);
+                latch.countDown();
+            }
+
+            @Override
+            public void addFailed(ManagedLedgerException exception, Object ctx) {
+                latch.countDown();
+            }
+        }, null);
+
+        latch.await();
+
+        assertTrue(addSuccess.get());
+
+        setFieldValue(ManagedLedgerImpl.class, ledger, "currentLedger", null);
+    }
+
+    private void setFieldValue(Class clazz, Object classObj, String fieldName, Object fieldValue)
throws Exception {
+        Field field = clazz.getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(classObj, fieldValue);
+    }
 }
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 9625987..048aecd 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -757,14 +757,17 @@ public class ServiceConfiguration implements PulsarConfiguration {
         doc = "operation timeout while updating managed-ledger metadata."
     )
     private long managedLedgerMetadataOperationsTimeoutSeconds = 60;
+
     @FieldContext(
             category = CATEGORY_STORAGE_ML,
             doc = "Read entries timeout when broker tries to read messages from bookkeeper
"
-                    + "(disable timeout by setting readTimeoutSeconds <= 0)"
+                    + "(0 to disable it)"
         )
-    private long managedLedgerReadEntryTimeoutSeconds = 60;
+    private long managedLedgerReadEntryTimeoutSeconds = 120;
         
-    
+    @FieldContext(category = CATEGORY_STORAGE_ML, 
+            doc = "Add entry timeout when broker tries to publish message to bookkeeper.(0
to disable it)")
+    private long managedLedgerAddEntryTimeoutSeconds = 120;
 
     /*** --- Load balancer --- ****/
     @FieldContext(
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index b2b5103..ff2a8dc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -745,6 +745,7 @@ public class BrokerService implements Closeable, ZooKeeperCacheListener<Policies
             managedLedgerConfig.setMetadataOperationsTimeoutSeconds(
                     serviceConfig.getManagedLedgerMetadataOperationsTimeoutSeconds());
             managedLedgerConfig.setReadEntryTimeoutSeconds(serviceConfig.getManagedLedgerReadEntryTimeoutSeconds());
+            managedLedgerConfig.setAddEntryTimeoutSeconds(serviceConfig.getManagedLedgerAddEntryTimeoutSeconds());
             managedLedgerConfig.setMetadataEnsembleSize(serviceConfig.getManagedLedgerDefaultEnsembleSize());
             managedLedgerConfig.setMetadataWriteQuorumSize(serviceConfig.getManagedLedgerDefaultWriteQuorum());
             managedLedgerConfig.setMetadataAckQuorumSize(serviceConfig.getManagedLedgerDefaultAckQuorum());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index e0130d2..0720d3e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -25,6 +25,7 @@ import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.net.URI;
 import java.net.URL;
 import java.util.ArrayList;
@@ -290,5 +291,11 @@ public abstract class MockedPulsarServiceBaseTest {
         }
     }
 
+    public static void setFieldValue(Class clazz, Object classObj, String fieldName, Object
fieldValue) throws Exception {
+        Field field = clazz.getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(classObj, fieldValue);
+    }
+    
     private static final Logger log = LoggerFactory.getLogger(MockedPulsarServiceBaseTest.class);
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
index 6988863..feb5b9e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
@@ -23,9 +23,12 @@ import static org.apache.pulsar.broker.service.BrokerService.BROKER_SERVICE_CONF
 import static org.mockito.Matchers.anyObject;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
@@ -33,6 +36,7 @@ import static org.testng.Assert.fail;
 
 import java.lang.reflect.Field;
 import java.net.URI;
+import java.security.GeneralSecurityException;
 import java.util.ArrayList;
 import java.util.IdentityHashMap;
 import java.util.List;
@@ -46,16 +50,24 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.client.PulsarMockBookKeeper;
+import org.apache.bookkeeper.client.PulsarMockLedgerHandle;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.namespace.OwnershipCache;
 import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.Message;
-import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.ProducerConsumerBase;
 import org.apache.pulsar.client.api.PulsarClient;
@@ -757,4 +769,66 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase
{
         }
     }
 
+    @Test(timeOut = 20000)
+    public void testAddEntryOperationTimeout() throws Exception {
+
+        log.info("-- Starting {} test --", methodName);
+
+        conf.setManagedLedgerAddEntryTimeoutSeconds(1);
+
+        final String topicName = "persistent://my-property/my-ns/addEntryTimeoutTopic";
+
+        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName).create();
+        ConsumerImpl<byte[]> consumer = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(topicName)
+                .subscriptionName("my-subscriber-name").subscribe();
+
+        PersistentTopic topic = (PersistentTopic) pulsar.getBrokerService().getOrCreateTopic(topicName).get();
+        ManagedLedgerImpl ml = (ManagedLedgerImpl) topic.getManagedLedger();
+
+        class MockLedgerHandle extends PulsarMockLedgerHandle {
+            public MockLedgerHandle(PulsarMockBookKeeper bk, long id, DigestType digest,
byte[] passwd)
+                    throws GeneralSecurityException {
+                super(bk, id, digest, passwd);
+            }
+
+            @Override
+            public void asyncAddEntry(final byte[] data, final AddCallback cb, final Object
ctx) {
+                // do nothing
+            }
+
+            @Override
+            public void asyncClose(org.apache.bookkeeper.client.AsyncCallback.CloseCallback
cb, Object ctx) {
+                cb.closeComplete(BKException.Code.OK, this, ctx);
+            }
+        }
+        MockLedgerHandle ledgerHandle = mock(MockLedgerHandle.class);
+        final byte[] data = "data".getBytes();
+        // this will make first entry to be timed out but then managed-ledger will create
a new ledger and next time add
+        // entry should be successful.
+        doNothing().when(ledgerHandle).asyncAddEntry(data, null, null);
+
+        MockedPulsarServiceBaseTest.setFieldValue(ManagedLedgerImpl.class, ml, "currentLedger",
ledgerHandle);
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicBoolean addedSuccessfully = new AtomicBoolean(false);
+        producer.sendAsync(data).handle((res, ex) -> {
+            if (ex == null) {
+                addedSuccessfully.set(true);
+            } else {
+                log.error("add-entry failed for {}", methodName, ex);
+            }
+            latch.countDown();
+            return null;
+        });
+        latch.await();
+
+        // broker should be resilient enough to add-entry timeout and add entry successfully.
+        assertTrue(addedSuccessfully.get());
+
+        byte[] receivedData = consumer.receive().getData();
+        assertEquals(receivedData, data);
+
+        producer.close();
+        consumer.close();
+    }
+    
 }
diff --git a/site/_data/config/broker.yaml b/site/_data/config/broker.yaml
index bbb3383..616873b 100644
--- a/site/_data/config/broker.yaml
+++ b/site/_data/config/broker.yaml
@@ -245,6 +245,9 @@ configs:
 - name: managedLedgerReadEntryTimeoutSeconds
   default: '120'
   description: Read entries timeout when broker tries to read messages from bookkeeper.
+- name: managedLedgerAddEntryTimeoutSeconds
+  default: '120'
+  description: Add entry timeout when broker tries to publish message to bookkeeper (0 to
disable it).
 - name: loadBalancerEnabled
   default: 'true'
   description: Enable load balancer


Mime
View raw message