Fix compile issue caused by Pulsar refactored Backoff implementation (#1793)

Author: Technoboy-

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchangeReplicator.java

@@ -16,6 +16,7 @@
 import static org.apache.pulsar.broker.service.persistent.PersistentTopic.MESSAGE_RATE_BACKOFF_MS;
 
 import io.streamnative.pulsar.handlers.amqp.impl.PersistentExchange;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -55,8 +56,11 @@ public abstract class AmqpExchangeReplicator implements AsyncCallbacks.ReadEntri
     private ManagedCursor cursor;
     private ScheduledExecutorService scheduledExecutorService;
 
-    protected final Backoff backOff = new Backoff(
-            100, TimeUnit.MILLISECONDS, 1, TimeUnit.MINUTES, 0, TimeUnit.MILLISECONDS);
+    protected final Backoff backOff = Backoff.builder()
+             .initialDelay(Duration.ZERO.ofMillis(100))
+                .maxBackoff(Duration.ofMinutes(1))
+            .mandatoryStop(Duration.ofMillis(0))
+            .build();
 
     private static final AtomicReferenceFieldUpdater<AmqpExchangeReplicator, State> STATE_UPDATER =
             AtomicReferenceFieldUpdater.newUpdater(AmqpExchangeReplicator.class, State.class, "state");
@@ -83,8 +87,11 @@ protected enum State {
             AtomicIntegerFieldUpdater.newUpdater(AmqpExchangeReplicator.class, "havePendingRead");
     private volatile int havePendingRead = FALSE;
 
-    private final Backoff readFailureBackoff = new Backoff(
-            1, TimeUnit.SECONDS, 1, TimeUnit.MINUTES, 0, TimeUnit.MILLISECONDS);
+    private final Backoff readFailureBackoff = Backoff.builder()
+            .initialDelay(Duration.ofSeconds(1))
+            .maxBackoff(Duration.ofMinutes(1))
+            .mandatoryStop(Duration.ofMillis(0))
+            .build();
 
     private final ExecutorService routeExecutor;
 
@@ -114,7 +121,7 @@ private void initMaxRouteQueueSize(int maxRouteQueueSize) {
 
     public void startReplicate() {
         if (STATE_UPDATER.get(AmqpExchangeReplicator.this).equals(AmqpExchangeReplicator.State.Stopping)) {
-            long waitTimeMs = backOff.next();
+            long waitTimeMs = backOff.next().toMillis();
             if (log.isDebugEnabled()) {
                 log.debug("{} Waiting for producer close before attempting reconnect, retrying in {} s",
                         name, waitTimeMs / 1000);
@@ -153,7 +160,7 @@ public void openCursorFailed(ManagedLedgerException e, Object o) {
 
     private void retryStartReplicator(Throwable ex) {
         if (STATE_UPDATER.compareAndSet(this, State.Starting, State.Stopped)) {
-            long waitTimeMs = backOff.next();
+            long waitTimeMs = backOff.next().toMillis();
             if (log.isDebugEnabled()) {
                 log.debug("{} Failed to start replicator, errorMsg: {}, retrying in {} s.",
                         name, ex.getMessage(), waitTimeMs / 1000);
@@ -242,7 +249,7 @@ public void readEntriesComplete(List<Entry> list, Object o) {
         }
         HAVE_PENDING_READ_UPDATER.set(this, FALSE);
         if (CollectionUtils.isEmpty(list)) {
-            long delay = readFailureBackoff.next();
+            long delay = readFailureBackoff.next().toMillis();
             log.warn("{} The read entry list is empty, will retry in {} ms. ReadPosition: {}, LAC: {}.",
                     name, delay, cursor.getReadPosition(), topic.getManagedLedger().getLastConfirmedEntry());
             scheduledExecutorService.schedule(this::readMoreEntries, delay, TimeUnit.MILLISECONDS);
@@ -329,7 +336,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object o) {
             return;
         }
 
-        long waitTimeMs = readFailureBackoff.next();
+        long waitTimeMs = readFailureBackoff.next().toMillis();
         if (log.isDebugEnabled()) {
             log.debug("{} Read entries from bookie failed, retrying in {} s", name, waitTimeMs / 1000, exception);
         }
@@ -380,7 +387,7 @@ public void closeFailed(ManagedLedgerException e, Object o) {
                         STATE_UPDATER.set(AmqpExchangeReplicator.this, State.Stopped);
                         return;
                     }
-                    long waitTimeMs = backOff.next();
+                    long waitTimeMs = backOff.next().toMillis();
                     log.error("[{}] AMQP Exchange Replicator stop failed. retrying in {} s",
                             name, waitTimeMs / 1000, e);
                     AmqpExchangeReplicator.this.scheduledExecutorService.schedule(

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpPulsarConsumer.java

@@ -14,6 +14,7 @@
 package io.streamnative.pulsar.handlers.amqp;
 
 import io.streamnative.pulsar.handlers.amqp.utils.MessageConvertUtils;
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -25,7 +26,6 @@
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.common.util.Backoff;
-import org.apache.pulsar.common.util.BackoffBuilder;
 import org.apache.qpid.server.protocol.v0_8.AMQShortString;
 
 /**
@@ -49,11 +49,11 @@ public AmqpPulsarConsumer(String consumerTag, Consumer<byte[]> consumer, boolean
         this.autoAck = autoAck;
         this.amqpChannel = amqpChannel;
         this.executorService = executorService;
-        this.consumeBackoff = new BackoffBuilder()
-                .setInitialTime(1, TimeUnit.MILLISECONDS)
-                .setMax(1, TimeUnit.SECONDS)
-                .setMandatoryStop(0, TimeUnit.SECONDS)
-                .create();
+        this.consumeBackoff = Backoff.builder()
+                .initialDelay(Duration.ofMillis(1))
+                .maxBackoff(Duration.ofSeconds(1))
+                .mandatoryStop(Duration.ofSeconds(0))
+                .build();
     }
 
     public void startConsume() {
@@ -69,7 +69,7 @@ private void consume() {
         try {
             message = this.consumer.receive(0, TimeUnit.SECONDS);
             if (message == null) {
-                this.executorService.schedule(this::consume, consumeBackoff.next(), TimeUnit.MILLISECONDS);
+                this.executorService.schedule(this::consume, consumeBackoff.next().toMillis(), TimeUnit.MILLISECONDS);
                 return;
             }
 
@@ -95,7 +95,7 @@ private void consume() {
             consumeBackoff.reset();
             this.consume();
         } catch (Exception e) {
-            long backoff = consumeBackoff.next();
+            long backoff = consumeBackoff.next().toMillis();
             log.error("Failed to receive message and send to client, retry in {} ms.", backoff, e);
             this.executorService.schedule(this::consume, backoff, TimeUnit.MILLISECONDS);
         }