From 9283a8e9d9692e20c0cac7d4b19d27ec7fcf06f1 Mon Sep 17 00:00:00 2001 From: Denovo1998 Date: Mon, 29 Jun 2026 19:23:01 +0800 Subject: [PATCH 1/6] [fix][broker] Handle synchronous schema lookup failures in replication --- .../persistent/GeoPersistentReplicator.java | 18 +- .../GeoPersistentReplicatorTest.java | 176 ++++++++++++++++++ 2 files changed, 193 insertions(+), 1 deletion(-) create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java index 84e936256a868..d3c0c6972be06 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java @@ -256,7 +256,23 @@ protected boolean replicateEntries(List entries, final InFlightTask inFli headersAndPayload.retain(); - CompletableFuture schemaFuture = getSchemaInfo(msg); + CompletableFuture schemaFuture; + try { + schemaFuture = getSchemaInfo(msg); + } catch (Exception e) { + log.warn() + .attr("position", entry.getPosition()) + .exception(e) + .log("Failed to get schema from local cluster, will try in the next loop"); + beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Fetching_Schema); + inFlightTask.incCompletedEntries(); + entry.release(); + headersAndPayload.release(); + msg.recycle(); + skipRemainingMessages = true; + doRewindCursor(false); + continue; + } if (!schemaFuture.isDone() || schemaFuture.isCompletedExceptionally()) { /** * Skip in flight reading tasks. diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java new file mode 100644 index 0000000000000..fb2b8719272a5 --- /dev/null +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java @@ -0,0 +1,176 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.broker.service.persistent; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Answers.RETURNS_SELF; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import org.apache.bookkeeper.mledger.Entry; +import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.mledger.Position; +import org.apache.bookkeeper.mledger.PositionFactory; +import org.apache.pulsar.broker.PulsarServerException; +import org.apache.pulsar.broker.PulsarService; +import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.service.BrokerService; +import org.apache.pulsar.broker.service.persistent.PersistentReplicator.InFlightTask; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.ProducerBuilder; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.MessageImpl; +import org.apache.pulsar.client.impl.ProducerImpl; +import org.apache.pulsar.client.impl.PulsarClientImpl; +import org.apache.pulsar.common.api.proto.MessageMetadata; +import org.apache.pulsar.common.protocol.Commands; +import org.apache.pulsar.common.schema.SchemaInfo; +import org.testng.annotations.Test; + +@Test(groups = "broker-replication") +public class GeoPersistentReplicatorTest { + + @Test + public void testSchemaInfoSynchronousFailureCompletesCurrentEntry() throws Exception { + ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(); + replicator.forceStarted(); + + Position position = PositionFactory.create(1, 2); + ByteBuf headersAndPayload = newMessageWithSchemaVersion(); + Entry entry = mock(Entry.class); + when(entry.getLength()).thenReturn(headersAndPayload.readableBytes()); + when(entry.getDataBuffer()).thenReturn(headersAndPayload); + when(entry.getPosition()).thenReturn(position); + when(entry.getLedgerId()).thenReturn(position.getLedgerId()); + when(entry.getEntryId()).thenReturn(position.getEntryId()); + doAnswer(invocation -> { + headersAndPayload.release(); + return null; + }).when(entry).release(); + + List entries = List.of(entry); + InFlightTask inFlightTask = new InFlightTask(position, 1, replicator.getReplicatorId()); + inFlightTask.setEntries(entries); + + try { + assertThat(replicator.replicateEntries(entries, inFlightTask)).isFalse(); + + assertThat(inFlightTask.getCompletedEntries()) + .as("the failed entry should release its in-flight permit") + .isEqualTo(1); + verify(entry).release(); + } finally { + while (headersAndPayload.refCnt() > 0) { + headersAndPayload.release(); + } + } + } + + private static ByteBuf newMessageWithSchemaVersion() { + MessageMetadata metadata = new MessageMetadata() + .setProducerName("producer") + .setSequenceId(1) + .setPublishTime(System.currentTimeMillis()) + .setSchemaVersion(new byte[] { 1 }); + ByteBuf payload = Unpooled.wrappedBuffer(new byte[] { 1 }); + try { + return Commands.serializeMetadataAndPayload(Commands.ChecksumType.Crc32c, metadata, payload); + } finally { + payload.release(); + } + } + + private static class ThrowingSchemaReplicator extends GeoPersistentReplicator { + + @SuppressWarnings("unchecked") + ThrowingSchemaReplicator() throws PulsarServerException { + this(mockTopic(), mockCursor(), "local", "remote", + mockBrokerService(), mockReplicationClient(), mock(PulsarAdmin.class)); + } + + private ThrowingSchemaReplicator(PersistentTopic topic, ManagedCursor cursor, String localCluster, + String remoteCluster, BrokerService brokerService, + PulsarClientImpl replicationClient, PulsarAdmin replicationAdmin) + throws PulsarServerException { + super(topic, cursor, localCluster, remoteCluster, brokerService, replicationClient, replicationAdmin); + } + + @Override + protected void startProducer() { + // Avoid creating a real remote producer from the superclass constructor. + } + + @Override + protected CompletableFuture getSchemaInfo(MessageImpl msg) throws ExecutionException { + throw new ExecutionException(new RuntimeException("injected schema provider failure")); + } + + void forceStarted() { + STATE_UPDATER.set(this, State.Started); + this.producer = mock(ProducerImpl.class); + } + + private static PersistentTopic mockTopic() throws PulsarServerException { + PersistentTopic topic = mock(PersistentTopic.class); + BrokerService brokerService = mockBrokerService(); + when(topic.getName()).thenReturn("persistent://public/default/t1"); + when(topic.getBrokerService()).thenReturn(brokerService); + when(topic.getReplicatorPrefix()).thenReturn("pulsar.repl"); + when(topic.getReplicatorDispatchRate()).thenReturn(null); + return topic; + } + + private static ManagedCursor mockCursor() { + ManagedCursor cursor = mock(ManagedCursor.class); + when(cursor.getName()).thenReturn("pulsar.repl.remote"); + return cursor; + } + + private static BrokerService mockBrokerService() throws PulsarServerException { + ServiceConfiguration config = new ServiceConfiguration(); + PulsarService pulsar = mock(PulsarService.class); + BrokerService brokerService = mock(BrokerService.class); + PulsarClientImpl localClient = mock(PulsarClientImpl.class); + PulsarAdmin admin = mock(PulsarAdmin.class); + + when(pulsar.getConfiguration()).thenReturn(config); + when(pulsar.getConfig()).thenReturn(config); + when(pulsar.getClient()).thenReturn(localClient); + when(pulsar.getAdminClient()).thenReturn(admin); + when(brokerService.pulsar()).thenReturn(pulsar); + when(brokerService.getPulsar()).thenReturn(pulsar); + return brokerService; + } + + @SuppressWarnings("unchecked") + private static PulsarClientImpl mockReplicationClient() { + PulsarClientImpl replicationClient = mock(PulsarClientImpl.class); + ProducerBuilder producerBuilder = mock(ProducerBuilder.class, RETURNS_SELF); + when(replicationClient.newProducer(any(Schema.class))).thenReturn(producerBuilder); + return replicationClient; + } + } +} From ef55f817298ffa3dd5b0bacfe54f69ee3590cf5c Mon Sep 17 00:00:00 2001 From: Denovo1998 Date: Tue, 30 Jun 2026 21:17:52 +0800 Subject: [PATCH 2/6] add test for schema lookup failures in replication --- .../persistent/GeoPersistentReplicator.java | 28 +++-- .../GeoPersistentReplicatorTest.java | 110 +++++++++++++----- 2 files changed, 96 insertions(+), 42 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java index d3c0c6972be06..d09cb9f70bccb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java @@ -23,6 +23,8 @@ import io.netty.buffer.ByteBuf; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.pulsar.broker.PulsarServerException; @@ -259,19 +261,8 @@ protected boolean replicateEntries(List entries, final InFlightTask inFli CompletableFuture schemaFuture; try { schemaFuture = getSchemaInfo(msg); - } catch (Exception e) { - log.warn() - .attr("position", entry.getPosition()) - .exception(e) - .log("Failed to get schema from local cluster, will try in the next loop"); - beforeTerminateOrCursorRewinding(ReasonOfWaitForCursorRewinding.Fetching_Schema); - inFlightTask.incCompletedEntries(); - entry.release(); - headersAndPayload.release(); - msg.recycle(); - skipRemainingMessages = true; - doRewindCursor(false); - continue; + } catch (ExecutionException e) { + schemaFuture = CompletableFuture.failedFuture(e); } if (!schemaFuture.isDone() || schemaFuture.isCompletedExceptionally()) { /** @@ -295,11 +286,18 @@ protected boolean replicateEntries(List entries, final InFlightTask inFli schemaFuture.whenComplete((__, e) -> { if (e != null) { log.warn() + .attr("backoffMs", PersistentTopic.MESSAGE_RATE_BACKOFF_MS) .exception(e) .log("Failed to get schema from local cluster, will try in the next loop"); + topic.getBrokerService().executor().schedule(() -> { + log.info("Resume the data replication after the schema fetching done"); + doRewindCursor(true); + }, + PersistentTopic.MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS); + } else { + log.info("Resume the data replication after the schema fetching done"); + doRewindCursor(true); } - log.info("Resume the data replication after the schema fetching done"); - doRewindCursor(true); }); } else { msg.setSchemaInfoForReplicator(schemaFuture.get()); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java index fb2b8719272a5..1adc22df95d50 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java @@ -21,15 +21,21 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Answers.RETURNS_SELF; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.channel.EventLoopGroup; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.bookkeeper.mledger.Position; @@ -54,34 +60,38 @@ public class GeoPersistentReplicatorTest { @Test - public void testSchemaInfoSynchronousFailureCompletesCurrentEntry() throws Exception { + public void testSchemaInfoSynchronousFailureReleasesBatchAndSchedulesCursorRewind() throws Exception { ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(); replicator.forceStarted(); - Position position = PositionFactory.create(1, 2); + Position firstPosition = PositionFactory.create(1, 2); ByteBuf headersAndPayload = newMessageWithSchemaVersion(); - Entry entry = mock(Entry.class); - when(entry.getLength()).thenReturn(headersAndPayload.readableBytes()); - when(entry.getDataBuffer()).thenReturn(headersAndPayload); - when(entry.getPosition()).thenReturn(position); - when(entry.getLedgerId()).thenReturn(position.getLedgerId()); - when(entry.getEntryId()).thenReturn(position.getEntryId()); - doAnswer(invocation -> { - headersAndPayload.release(); - return null; - }).when(entry).release(); + Entry firstEntry = newEntry(firstPosition, headersAndPayload); + Entry secondEntry = mock(Entry.class); - List entries = List.of(entry); - InFlightTask inFlightTask = new InFlightTask(position, 1, replicator.getReplicatorId()); + List entries = List.of(firstEntry, secondEntry); + InFlightTask inFlightTask = new InFlightTask(firstPosition, entries.size(), replicator.getReplicatorId()); inFlightTask.setEntries(entries); try { assertThat(replicator.replicateEntries(entries, inFlightTask)).isFalse(); assertThat(inFlightTask.getCompletedEntries()) - .as("the failed entry should release its in-flight permit") - .isEqualTo(1); - verify(entry).release(); + .as("the failed entry and skipped remaining entries should release their in-flight permits") + .isEqualTo(entries.size()); + verify(firstEntry).release(); + verify(secondEntry).release(); + verify(replicator.cursor, never()).rewind(); + verify(replicator.context.executor).schedule(any(Runnable.class), + eq((long) PersistentTopic.MESSAGE_RATE_BACKOFF_MS), eq(TimeUnit.MILLISECONDS)); + + Runnable scheduledRewind = replicator.context.scheduledTask.get(); + assertThat(scheduledRewind).isNotNull(); + assertThat(replicator.readMoreEntriesCalls).isZero(); + + scheduledRewind.run(); + verify(replicator.cursor).rewind(); + assertThat(replicator.readMoreEntriesCalls).isEqualTo(1); } finally { while (headersAndPayload.refCnt() > 0) { headersAndPayload.release(); @@ -89,6 +99,20 @@ public void testSchemaInfoSynchronousFailureCompletesCurrentEntry() throws Excep } } + private static Entry newEntry(Position position, ByteBuf headersAndPayload) { + Entry entry = mock(Entry.class); + when(entry.getLength()).thenReturn(headersAndPayload.readableBytes()); + when(entry.getDataBuffer()).thenReturn(headersAndPayload); + when(entry.getPosition()).thenReturn(position); + when(entry.getLedgerId()).thenReturn(position.getLedgerId()); + when(entry.getEntryId()).thenReturn(position.getEntryId()); + doAnswer(invocation -> { + headersAndPayload.release(); + return null; + }).when(entry).release(); + return entry; + } + private static ByteBuf newMessageWithSchemaVersion() { MessageMetadata metadata = new MessageMetadata() .setProducerName("producer") @@ -105,17 +129,18 @@ private static ByteBuf newMessageWithSchemaVersion() { private static class ThrowingSchemaReplicator extends GeoPersistentReplicator { - @SuppressWarnings("unchecked") + private final ReplicatorContext context; + private int readMoreEntriesCalls; + ThrowingSchemaReplicator() throws PulsarServerException { - this(mockTopic(), mockCursor(), "local", "remote", - mockBrokerService(), mockReplicationClient(), mock(PulsarAdmin.class)); + this(new ReplicatorContext()); } - private ThrowingSchemaReplicator(PersistentTopic topic, ManagedCursor cursor, String localCluster, - String remoteCluster, BrokerService brokerService, - PulsarClientImpl replicationClient, PulsarAdmin replicationAdmin) + private ThrowingSchemaReplicator(ReplicatorContext context) throws PulsarServerException { - super(topic, cursor, localCluster, remoteCluster, brokerService, replicationClient, replicationAdmin); + super(context.topic, context.cursor, "local", "remote", context.brokerService, + context.replicationClient, context.replicationAdmin); + this.context = context; } @Override @@ -128,14 +153,18 @@ protected CompletableFuture getSchemaInfo(MessageImpl msg) throws Ex throw new ExecutionException(new RuntimeException("injected schema provider failure")); } + @Override + protected void readMoreEntries() { + readMoreEntriesCalls++; + } + void forceStarted() { STATE_UPDATER.set(this, State.Started); this.producer = mock(ProducerImpl.class); } - private static PersistentTopic mockTopic() throws PulsarServerException { + private static PersistentTopic mockTopic(BrokerService brokerService) { PersistentTopic topic = mock(PersistentTopic.class); - BrokerService brokerService = mockBrokerService(); when(topic.getName()).thenReturn("persistent://public/default/t1"); when(topic.getBrokerService()).thenReturn(brokerService); when(topic.getReplicatorPrefix()).thenReturn("pulsar.repl"); @@ -149,7 +178,9 @@ private static ManagedCursor mockCursor() { return cursor; } - private static BrokerService mockBrokerService() throws PulsarServerException { + private static BrokerService mockBrokerService(EventLoopGroup executor, + AtomicReference scheduledTask) + throws PulsarServerException { ServiceConfiguration config = new ServiceConfiguration(); PulsarService pulsar = mock(PulsarService.class); BrokerService brokerService = mock(BrokerService.class); @@ -162,6 +193,11 @@ private static BrokerService mockBrokerService() throws PulsarServerException { when(pulsar.getAdminClient()).thenReturn(admin); when(brokerService.pulsar()).thenReturn(pulsar); when(brokerService.getPulsar()).thenReturn(pulsar); + when(brokerService.executor()).thenReturn(executor); + doAnswer(invocation -> { + scheduledTask.set(invocation.getArgument(0, Runnable.class)); + return null; + }).when(executor).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class)); return brokerService; } @@ -173,4 +209,24 @@ private static PulsarClientImpl mockReplicationClient() { return replicationClient; } } + + private static class ReplicatorContext { + private final AtomicReference scheduledTask; + private final EventLoopGroup executor; + private final BrokerService brokerService; + private final PersistentTopic topic; + private final ManagedCursor cursor; + private final PulsarClientImpl replicationClient; + private final PulsarAdmin replicationAdmin; + + private ReplicatorContext() throws PulsarServerException { + this.scheduledTask = new AtomicReference<>(); + this.executor = mock(EventLoopGroup.class); + this.brokerService = ThrowingSchemaReplicator.mockBrokerService(executor, scheduledTask); + this.topic = ThrowingSchemaReplicator.mockTopic(brokerService); + this.cursor = ThrowingSchemaReplicator.mockCursor(); + this.replicationClient = ThrowingSchemaReplicator.mockReplicationClient(); + this.replicationAdmin = mock(PulsarAdmin.class); + } + } } From 4a549ae2c7424bba2100556088146b6e4e2ab872 Mon Sep 17 00:00:00 2001 From: Denovo1998 Date: Thu, 2 Jul 2026 20:37:46 +0800 Subject: [PATCH 3/6] handle synchronous schema lookup failures in replication --- .../broker/service/persistent/PersistentReplicator.java | 4 ++++ .../service/persistent/GeoPersistentReplicatorTest.java | 5 ++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java index e3d2ec7115006..2a9de2d4c5f2d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java @@ -433,6 +433,10 @@ public void readEntriesComplete(List entries, Object ctx) { .attr("atLeastOneMessageSentForReplication", atLeastOneMessageSentForReplication) .attr("isWritable", isWritable()) .log("Pausing replication traffic"); + } else if (waitForCursorRewindingRefCnf > 0) { + log.debug() + .attr("reason", reasonOfWaitForCursorRewinding) + .log("Skipping read while waiting for cursor rewind"); } else { readMoreEntries(); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java index 1adc22df95d50..731c6dde77866 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java @@ -60,7 +60,7 @@ public class GeoPersistentReplicatorTest { @Test - public void testSchemaInfoSynchronousFailureReleasesBatchAndSchedulesCursorRewind() throws Exception { + public void testSchemaInfoSynchronousFailureSkipsOuterReadUntilScheduledCursorRewind() throws Exception { ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(); replicator.forceStarted(); @@ -71,10 +71,9 @@ public void testSchemaInfoSynchronousFailureReleasesBatchAndSchedulesCursorRewin List entries = List.of(firstEntry, secondEntry); InFlightTask inFlightTask = new InFlightTask(firstPosition, entries.size(), replicator.getReplicatorId()); - inFlightTask.setEntries(entries); try { - assertThat(replicator.replicateEntries(entries, inFlightTask)).isFalse(); + replicator.readEntriesComplete(entries, inFlightTask); assertThat(inFlightTask.getCompletedEntries()) .as("the failed entry and skipped remaining entries should release their in-flight permits") From 7fc0f85c23fefdb4575b4136a78d1f8d2d44d0ed Mon Sep 17 00:00:00 2001 From: Denovo1998 Date: Sun, 26 Jul 2026 08:15:06 +0800 Subject: [PATCH 4/6] [fix][broker] Normalize synchronous schema lookup failures --- .../persistent/GeoPersistentReplicator.java | 8 +-- .../persistent/PersistentReplicator.java | 15 ++++-- .../GeoPersistentReplicatorTest.java | 51 +++++++++++++------ 3 files changed, 46 insertions(+), 28 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java index b724b16bd6438..0ece3d9be81d5 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java @@ -23,7 +23,6 @@ import io.netty.buffer.ByteBuf; import java.util.List; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; @@ -258,12 +257,7 @@ protected boolean replicateEntries(List entries, final InFlightTask inFli headersAndPayload.retain(); - CompletableFuture schemaFuture; - try { - schemaFuture = getSchemaInfo(msg); - } catch (ExecutionException e) { - schemaFuture = CompletableFuture.failedFuture(e); - } + CompletableFuture schemaFuture = getSchemaInfo(msg); if (!schemaFuture.isDone() || schemaFuture.isCompletedExceptionally()) { /** * Skip in flight reading tasks. diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java index e6ce7ee47fe2d..0763c265fdb1c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java @@ -26,6 +26,7 @@ import static org.apache.pulsar.broker.service.AbstractReplicator.State.Terminating; import static org.apache.pulsar.broker.service.persistent.PersistentTopic.MESSAGE_RATE_BACKOFF_MS; import com.google.common.annotations.VisibleForTesting; +import com.google.common.util.concurrent.ExecutionError; import io.github.merlimat.slog.Logger; import io.netty.buffer.ByteBuf; import io.netty.util.Recycler; @@ -466,12 +467,16 @@ public void readEntriesComplete(List entries, Object ctx) { protected abstract boolean replicateEntries(List entries, InFlightTask inFlightTask); - protected CompletableFuture getSchemaInfo(MessageImpl msg) throws ExecutionException { - if (msg.getSchemaVersion() == null || msg.getSchemaVersion().length == 0) { - return CompletableFuture.completedFuture(null); + protected CompletableFuture getSchemaInfo(MessageImpl msg) { + try { + if (msg.getSchemaVersion() == null || msg.getSchemaVersion().length == 0) { + return CompletableFuture.completedFuture(null); + } + return client.getSchemaProviderLoadingCache().get(localSchemaTopicName) + .getSchemaByVersion(msg.getSchemaVersion()); + } catch (ExecutionException | RuntimeException | ExecutionError e) { + return CompletableFuture.failedFuture(e); } - return client.getSchemaProviderLoadingCache().get(localSchemaTopicName) - .getSchemaByVersion(msg.getSchemaVersion()); } public void updateCursorState() { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java index 731c6dde77866..63881d97c48cd 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java @@ -22,17 +22,20 @@ import static org.mockito.Answers.RETURNS_SELF; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.google.common.cache.LoadingCache; +import com.google.common.util.concurrent.ExecutionError; +import com.google.common.util.concurrent.UncheckedExecutionException; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.EventLoopGroup; import java.util.List; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -48,20 +51,30 @@ import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.impl.MessageImpl; +import org.apache.pulsar.client.api.schema.SchemaInfoProvider; import org.apache.pulsar.client.impl.ProducerImpl; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.common.api.proto.MessageMetadata; import org.apache.pulsar.common.protocol.Commands; -import org.apache.pulsar.common.schema.SchemaInfo; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @Test(groups = "broker-replication") public class GeoPersistentReplicatorTest { - @Test - public void testSchemaInfoSynchronousFailureSkipsOuterReadUntilScheduledCursorRewind() throws Exception { - ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(); + @DataProvider + public static Object[][] synchronousSchemaLookupFailures() { + return new Object[][] { + { new ExecutionException(new RuntimeException("checked schema provider failure")) }, + { new UncheckedExecutionException(new RuntimeException("unchecked schema provider failure")) }, + { new ExecutionError(new AssertionError("schema provider error")) } + }; + } + + @Test(dataProvider = "synchronousSchemaLookupFailures") + public void testSchemaInfoSynchronousFailureSkipsOuterReadUntilScheduledCursorRewind( + Throwable schemaLookupFailure) throws Exception { + ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(schemaLookupFailure); replicator.forceStarted(); Position firstPosition = PositionFactory.create(1, 2); @@ -80,6 +93,9 @@ public void testSchemaInfoSynchronousFailureSkipsOuterReadUntilScheduledCursorRe .isEqualTo(entries.size()); verify(firstEntry).release(); verify(secondEntry).release(); + assertThat(headersAndPayload.refCnt()) + .as("the entry and retained schema lookup buffer should both be released") + .isZero(); verify(replicator.cursor, never()).rewind(); verify(replicator.context.executor).schedule(any(Runnable.class), eq((long) PersistentTopic.MESSAGE_RATE_BACKOFF_MS), eq(TimeUnit.MILLISECONDS)); @@ -131,8 +147,9 @@ private static class ThrowingSchemaReplicator extends GeoPersistentReplicator { private final ReplicatorContext context; private int readMoreEntriesCalls; - ThrowingSchemaReplicator() throws PulsarServerException { - this(new ReplicatorContext()); + ThrowingSchemaReplicator(Throwable schemaLookupFailure) + throws PulsarServerException, ExecutionException { + this(new ReplicatorContext(schemaLookupFailure)); } private ThrowingSchemaReplicator(ReplicatorContext context) @@ -147,11 +164,6 @@ protected void startProducer() { // Avoid creating a real remote producer from the superclass constructor. } - @Override - protected CompletableFuture getSchemaInfo(MessageImpl msg) throws ExecutionException { - throw new ExecutionException(new RuntimeException("injected schema provider failure")); - } - @Override protected void readMoreEntries() { readMoreEntriesCalls++; @@ -201,9 +213,15 @@ private static BrokerService mockBrokerService(EventLoopGroup executor, } @SuppressWarnings("unchecked") - private static PulsarClientImpl mockReplicationClient() { + private static PulsarClientImpl mockReplicationClient(Throwable schemaLookupFailure) + throws ExecutionException { PulsarClientImpl replicationClient = mock(PulsarClientImpl.class); + LoadingCache schemaProviderLoadingCache = mock(LoadingCache.class); ProducerBuilder producerBuilder = mock(ProducerBuilder.class, RETURNS_SELF); + when(replicationClient.getSchemaProviderLoadingCache()).thenReturn(schemaProviderLoadingCache); + doAnswer(__ -> { + throw schemaLookupFailure; + }).when(schemaProviderLoadingCache).get(anyString()); when(replicationClient.newProducer(any(Schema.class))).thenReturn(producerBuilder); return replicationClient; } @@ -218,13 +236,14 @@ private static class ReplicatorContext { private final PulsarClientImpl replicationClient; private final PulsarAdmin replicationAdmin; - private ReplicatorContext() throws PulsarServerException { + private ReplicatorContext(Throwable schemaLookupFailure) + throws PulsarServerException, ExecutionException { this.scheduledTask = new AtomicReference<>(); this.executor = mock(EventLoopGroup.class); this.brokerService = ThrowingSchemaReplicator.mockBrokerService(executor, scheduledTask); this.topic = ThrowingSchemaReplicator.mockTopic(brokerService); this.cursor = ThrowingSchemaReplicator.mockCursor(); - this.replicationClient = ThrowingSchemaReplicator.mockReplicationClient(); + this.replicationClient = ThrowingSchemaReplicator.mockReplicationClient(schemaLookupFailure); this.replicationAdmin = mock(PulsarAdmin.class); } } From a6b5d111489e5ae08c939fd1b20542a6d0d86e88 Mon Sep 17 00:00:00 2001 From: Denovo1998 Date: Mon, 3 Aug 2026 20:42:55 +0800 Subject: [PATCH 5/6] [fix][broker] Keep cursor rewind checks under in-flight lock --- .../persistent/PersistentReplicator.java | 4 --- .../GeoPersistentReplicatorTest.java | 35 +++++++++---------- 2 files changed, 17 insertions(+), 22 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java index 0763c265fdb1c..5009f156d7ff4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java @@ -456,10 +456,6 @@ public void readEntriesComplete(List entries, Object ctx) { .attr("atLeastOneMessageSentForReplication", atLeastOneMessageSentForReplication) .attr("isWritable", isWritable()) .log("Pausing replication traffic"); - } else if (waitForCursorRewindingRefCnf > 0) { - log.debug() - .attr("reason", reasonOfWaitForCursorRewinding) - .log("Skipping read while waiting for cursor rewind"); } else { readMoreEntries(); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java index 63881d97c48cd..c4b2558a0ca84 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java @@ -21,9 +21,11 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Answers.RETURNS_SELF; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -35,10 +37,10 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.EventLoopGroup; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; import org.apache.bookkeeper.mledger.Entry; import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.bookkeeper.mledger.Position; @@ -72,7 +74,7 @@ public static Object[][] synchronousSchemaLookupFailures() { } @Test(dataProvider = "synchronousSchemaLookupFailures") - public void testSchemaInfoSynchronousFailureSkipsOuterReadUntilScheduledCursorRewind( + public void testSchemaInfoSynchronousFailureDoesNotReadUntilScheduledCursorRewind( Throwable schemaLookupFailure) throws Exception { ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(schemaLookupFailure); replicator.forceStarted(); @@ -83,7 +85,7 @@ public void testSchemaInfoSynchronousFailureSkipsOuterReadUntilScheduledCursorRe Entry secondEntry = mock(Entry.class); List entries = List.of(firstEntry, secondEntry); - InFlightTask inFlightTask = new InFlightTask(firstPosition, entries.size(), replicator.getReplicatorId()); + InFlightTask inFlightTask = replicator.createOrRecycleInFlightTaskIntoQueue(firstPosition, entries.size()); try { replicator.readEntriesComplete(entries, inFlightTask); @@ -97,16 +99,18 @@ public void testSchemaInfoSynchronousFailureSkipsOuterReadUntilScheduledCursorRe .as("the entry and retained schema lookup buffer should both be released") .isZero(); verify(replicator.cursor, never()).rewind(); - verify(replicator.context.executor).schedule(any(Runnable.class), + verify(replicator.cursor, never()).asyncReadEntriesOrWait( + anyInt(), anyLong(), any(), any(), any()); + verify(replicator.context.executor, atLeastOnce()).schedule(any(Runnable.class), eq((long) PersistentTopic.MESSAGE_RATE_BACKOFF_MS), eq(TimeUnit.MILLISECONDS)); - Runnable scheduledRewind = replicator.context.scheduledTask.get(); + Runnable scheduledRewind = replicator.context.scheduledTasks.get(0); assertThat(scheduledRewind).isNotNull(); - assertThat(replicator.readMoreEntriesCalls).isZero(); scheduledRewind.run(); verify(replicator.cursor).rewind(); - assertThat(replicator.readMoreEntriesCalls).isEqualTo(1); + verify(replicator.cursor).asyncReadEntriesOrWait( + anyInt(), anyLong(), any(), any(), any()); } finally { while (headersAndPayload.refCnt() > 0) { headersAndPayload.release(); @@ -145,7 +149,6 @@ private static ByteBuf newMessageWithSchemaVersion() { private static class ThrowingSchemaReplicator extends GeoPersistentReplicator { private final ReplicatorContext context; - private int readMoreEntriesCalls; ThrowingSchemaReplicator(Throwable schemaLookupFailure) throws PulsarServerException, ExecutionException { @@ -164,11 +167,6 @@ protected void startProducer() { // Avoid creating a real remote producer from the superclass constructor. } - @Override - protected void readMoreEntries() { - readMoreEntriesCalls++; - } - void forceStarted() { STATE_UPDATER.set(this, State.Started); this.producer = mock(ProducerImpl.class); @@ -186,11 +184,12 @@ private static PersistentTopic mockTopic(BrokerService brokerService) { private static ManagedCursor mockCursor() { ManagedCursor cursor = mock(ManagedCursor.class); when(cursor.getName()).thenReturn("pulsar.repl.remote"); + when(cursor.getReadPosition()).thenReturn(PositionFactory.create(1, 1)); return cursor; } private static BrokerService mockBrokerService(EventLoopGroup executor, - AtomicReference scheduledTask) + List scheduledTasks) throws PulsarServerException { ServiceConfiguration config = new ServiceConfiguration(); PulsarService pulsar = mock(PulsarService.class); @@ -206,7 +205,7 @@ private static BrokerService mockBrokerService(EventLoopGroup executor, when(brokerService.getPulsar()).thenReturn(pulsar); when(brokerService.executor()).thenReturn(executor); doAnswer(invocation -> { - scheduledTask.set(invocation.getArgument(0, Runnable.class)); + scheduledTasks.add(invocation.getArgument(0, Runnable.class)); return null; }).when(executor).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class)); return brokerService; @@ -228,7 +227,7 @@ private static PulsarClientImpl mockReplicationClient(Throwable schemaLookupFail } private static class ReplicatorContext { - private final AtomicReference scheduledTask; + private final List scheduledTasks; private final EventLoopGroup executor; private final BrokerService brokerService; private final PersistentTopic topic; @@ -238,9 +237,9 @@ private static class ReplicatorContext { private ReplicatorContext(Throwable schemaLookupFailure) throws PulsarServerException, ExecutionException { - this.scheduledTask = new AtomicReference<>(); + this.scheduledTasks = new ArrayList<>(); this.executor = mock(EventLoopGroup.class); - this.brokerService = ThrowingSchemaReplicator.mockBrokerService(executor, scheduledTask); + this.brokerService = ThrowingSchemaReplicator.mockBrokerService(executor, scheduledTasks); this.topic = ThrowingSchemaReplicator.mockTopic(brokerService); this.cursor = ThrowingSchemaReplicator.mockCursor(); this.replicationClient = ThrowingSchemaReplicator.mockReplicationClient(schemaLookupFailure); From b608981b758dfb238e4708f2672069aad30dc913 Mon Sep 17 00:00:00 2001 From: Denovo1998 Date: Sat, 29 Aug 2026 10:14:52 +0800 Subject: [PATCH 6/6] [fix][broker] Handle synchronous schema lookup failures in replication[fix][broker] Handle synchronous schema lookup failures in replication --- .../broker/service/persistent/GeoPersistentReplicator.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java index 0ece3d9be81d5..df71f28cb4957 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicator.java @@ -282,9 +282,9 @@ protected boolean replicateEntries(List entries, final InFlightTask inFli log.warn() .attr("backoffMs", PersistentTopic.MESSAGE_RATE_BACKOFF_MS) .exception(e) - .log("Failed to get schema from local cluster, will try in the next loop"); + .log("Failed to get schema from local cluster, will retry after backoff"); topic.getBrokerService().executor().schedule(() -> { - log.info("Resume the data replication after the schema fetching done"); + log.debug("Resume the data replication after the schema fetching done"); doRewindCursor(true); }, PersistentTopic.MESSAGE_RATE_BACKOFF_MS, TimeUnit.MILLISECONDS);