diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java index 5343e2d0e330d..9f3bd48f5e758 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsController.java @@ -41,6 +41,7 @@ import org.apache.bookkeeper.mledger.PositionFactory; import org.apache.commons.lang3.mutable.MutableBoolean; import org.apache.pulsar.broker.service.Replicator; +import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.stats.OpenTelemetryReplicatedSubscriptionStats; import org.apache.pulsar.common.api.proto.ClusterMessageId; @@ -52,6 +53,7 @@ import org.apache.pulsar.common.api.proto.ReplicatedSubscriptionsSnapshotResponse; import org.apache.pulsar.common.api.proto.ReplicatedSubscriptionsUpdate; import org.apache.pulsar.common.protocol.Markers; +import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.opentelemetry.annotations.PulsarDeprecatedMetric; /** @@ -214,7 +216,7 @@ private void receiveSubscriptionUpdated(ReplicatedSubscriptionsUpdate update) { PersistentSubscription sub = topic.getSubscription(update.getSubscriptionName()); if (sub != null) { - sub.acknowledgeMessageAsync(Collections.singletonList(pos), AckType.Cumulative, Collections.emptyMap()); + acknowledgeSubscriptionUpdate(update.getSubscriptionName(), sub, pos); } else { // Subscription doesn't exist. We need to force the creation of the subscription in this cluster. log.info() @@ -224,13 +226,35 @@ private void receiveSubscriptionUpdated(ReplicatedSubscriptionsUpdate update) { .log("Creating subscription at: after receiving update from replicated subscription"); topic.createSubscription(update.getSubscriptionName(), InitialPosition.Earliest, true /* replicateSubscriptionState */, Collections.emptyMap()) - .thenAccept(subscriptionCreated -> { - subscriptionCreated.acknowledgeMessageAsync(Collections.singletonList(pos), - AckType.Cumulative, Collections.emptyMap()); + .thenAccept(subscriptionCreated -> + acknowledgeSubscriptionUpdate(update.getSubscriptionName(), subscriptionCreated, pos)) + .exceptionally(e -> { + if (e != null) { + log.warn() + .attr("subscriptionName", update.getSubscriptionName()) + .attr("pos", pos) + .exception(FutureUtil.unwrapCompletionException(e)) + .log("Failed to create replicated subscription"); + } + return null; }); } } + private CompletableFuture acknowledgeSubscriptionUpdate(String subscriptionName, Subscription sub, + Position pos) { + return sub.acknowledgeMessageAsync(Collections.singletonList(pos), AckType.Cumulative, Collections.emptyMap()) + .whenComplete((__, e) -> { + if (e != null) { + log.warn() + .attr("subscriptionName", subscriptionName) + .attr("pos", pos) + .exception(FutureUtil.unwrapCompletionException(e)) + .log("Failed to update replicated subscription"); + } + }); + } + private void startNewSnapshot() { cleanupTimedOutSnapshots(); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java index ab05790f9a40c..80c4986896eac 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionsControllerTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.service.persistent; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; @@ -42,6 +43,9 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; +import java.util.function.BiFunction; +import java.util.function.Function; import lombok.Cleanup; import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback; import org.apache.bookkeeper.mledger.ManagedLedger; @@ -54,7 +58,10 @@ import org.apache.pulsar.broker.service.Replicator; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.stats.OpenTelemetryReplicatedSubscriptionStats; +import org.apache.pulsar.common.api.proto.CommandAck.AckType; +import org.apache.pulsar.common.api.proto.CommandSubscribe.InitialPosition; import org.apache.pulsar.common.api.proto.MarkerType; +import org.apache.pulsar.common.api.proto.MarkersMessageIdData; import org.apache.pulsar.common.policies.data.BacklogQuota; import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl; import org.apache.pulsar.common.protocol.Commands; @@ -66,6 +73,100 @@ @Test(groups = "broker-replication") public class ReplicatedSubscriptionsControllerTest { + @Test + @SuppressWarnings({"rawtypes", "unchecked"}) + public void testReplicatedSubscriptionUpdateAckFutureIsObserved() throws Exception { + PulsarService pulsar = mock(PulsarService.class); + ScheduledExecutorService executor = mock(ScheduledExecutorService.class); + ScheduledFuture timer = mock(ScheduledFuture.class); + ServiceConfiguration config = new ServiceConfiguration(); + config.setEnableReplicatedSubscriptions(true); + OpenTelemetryReplicatedSubscriptionStats stats = mock(OpenTelemetryReplicatedSubscriptionStats.class); + BrokerService brokerService = mock(BrokerService.class); + PersistentTopic topic = mock(PersistentTopic.class); + PersistentSubscription subscription = mock(PersistentSubscription.class); + ObservableFuture ackFuture = new ObservableFuture<>(); + + when(brokerService.pulsar()).thenReturn(pulsar); + when(brokerService.getPulsar()).thenReturn(pulsar); + when(pulsar.getExecutor()).thenReturn(executor); + when(pulsar.getConfiguration()).thenReturn(config); + when(pulsar.getOpenTelemetryReplicatedSubscriptionStats()).thenReturn(stats); + when(executor.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class))) + .thenReturn(timer); + + when(topic.getName()).thenReturn("persistent://public/default/t1"); + when(topic.getBrokerService()).thenReturn(brokerService); + when(topic.getSubscription("sub")).thenReturn(subscription); + when(subscription.acknowledgeMessageAsync(any(), eq(AckType.Cumulative), any())).thenReturn(ackFuture); + + ReplicatedSubscriptionsController controller = new ReplicatedSubscriptionsController(topic, "local"); + ByteBuf marker = Markers.newReplicatedSubscriptionsUpdate("sub", + Map.of("local", new MarkersMessageIdData().setLedgerId(1).setEntryId(2))); + try { + Commands.skipMessageMetadata(marker); + + controller.receivedReplicatedSubscriptionMarker(PositionFactory.create(3, 4), + MarkerType.REPLICATED_SUBSCRIPTION_UPDATE_VALUE, marker); + + assertThat(ackFuture.isObserved()) + .as("replicated subscription update ack failures should not be dropped") + .isTrue(); + } finally { + marker.release(); + controller.close(); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testReplicatedSubscriptionUpdateCreatedSubscriptionAckFutureIsObserved() throws Exception { + PulsarService pulsar = mock(PulsarService.class); + ScheduledExecutorService executor = mock(ScheduledExecutorService.class); + @SuppressWarnings("rawtypes") + ScheduledFuture timer = mock(ScheduledFuture.class); + ServiceConfiguration config = new ServiceConfiguration(); + config.setEnableReplicatedSubscriptions(true); + OpenTelemetryReplicatedSubscriptionStats stats = mock(OpenTelemetryReplicatedSubscriptionStats.class); + BrokerService brokerService = mock(BrokerService.class); + PersistentTopic topic = mock(PersistentTopic.class); + PersistentSubscription createdSubscription = mock(PersistentSubscription.class); + ObservableFuture ackFuture = new ObservableFuture<>(); + + when(brokerService.pulsar()).thenReturn(pulsar); + when(brokerService.getPulsar()).thenReturn(pulsar); + when(pulsar.getExecutor()).thenReturn(executor); + when(pulsar.getConfiguration()).thenReturn(config); + when(pulsar.getOpenTelemetryReplicatedSubscriptionStats()).thenReturn(stats); + when(executor.scheduleAtFixedRate(any(Runnable.class), anyLong(), anyLong(), any(TimeUnit.class))) + .thenReturn(timer); + + when(topic.getName()).thenReturn("persistent://public/default/t1"); + when(topic.getBrokerService()).thenReturn(brokerService); + when(topic.getSubscription("sub")).thenReturn(null); + when(topic.createSubscription(eq("sub"), eq(InitialPosition.Earliest), eq(true), any())) + .thenReturn(CompletableFuture.completedFuture(createdSubscription)); + when(createdSubscription.acknowledgeMessageAsync(any(), eq(AckType.Cumulative), any())).thenReturn(ackFuture); + + ReplicatedSubscriptionsController controller = new ReplicatedSubscriptionsController(topic, "local"); + ByteBuf marker = Markers.newReplicatedSubscriptionsUpdate("sub", + Map.of("local", new MarkersMessageIdData().setLedgerId(1).setEntryId(2))); + try { + Commands.skipMessageMetadata(marker); + + controller.receivedReplicatedSubscriptionMarker(PositionFactory.create(3, 4), + MarkerType.REPLICATED_SUBSCRIPTION_UPDATE_VALUE, marker); + + verify(topic).createSubscription(eq("sub"), eq(InitialPosition.Earliest), eq(true), any()); + assertThat(ackFuture.isObserved()) + .as("replicated subscription update ack failures should be observed after subscription creation") + .isTrue(); + } finally { + marker.release(); + controller.close(); + } + } + @Test @SuppressWarnings("unchecked") public void testFinalSnapshotMarkerPublishFailureKeepsSnapshotPending() { @@ -306,4 +407,40 @@ public Set> entrySet() { controller.close(); } } + + private static class ObservableFuture extends CompletableFuture { + private boolean observed; + + boolean isObserved() { + return observed; + } + + private void markObserved() { + observed = true; + } + + @Override + public CompletableFuture whenComplete(BiConsumer action) { + markObserved(); + return super.whenComplete(action); + } + + @Override + public CompletableFuture whenCompleteAsync(BiConsumer action) { + markObserved(); + return super.whenCompleteAsync(action); + } + + @Override + public CompletableFuture exceptionally(Function fn) { + markObserved(); + return super.exceptionally(fn); + } + + @Override + public CompletableFuture handle(BiFunction fn) { + markObserved(); + return super.handle(fn); + } + } }