You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/16 22:31:30 UTC
(pulsar) 07/24: [fix][broker] Handle synchronous schema lookup failures in replication (#26108)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0-backports-2026-10-17
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a2c35f624b31b27f044513a3c5508f542f23b207
Author: sinan liu <li...@gmail.com>
AuthorDate: Thu Oct 15 10:35:04 2026 +0800
[fix][broker] Handle synchronous schema lookup failures in replication (#26108)
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
(cherry picked from commit 43e0d588d231af3172173a9d1cf73ab97595751c)
---
.../persistent/GeoPersistentReplicator.java | 6 +-
.../service/persistent/PersistentReplicator.java | 35 ++-
.../persistent/GeoPersistentReplicatorTest.java | 275 +++++++++++++++++++++
3 files changed, 309 insertions(+), 7 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 9027aa41ac44..aab2102af9d2 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
@@ -213,9 +213,11 @@ public class GeoPersistentReplicator extends PersistentReplicator {
if (e != null) {
log.warn("[{}] Failed to get schema from local cluster, will try in the next loop",
replicatorId, e);
+ doRewindCursorAndDelayReadRetry();
+ } else {
+ log.info("[{}] Resume the data replication after the schema fetching done", replicatorId);
+ doRewindCursor(true);
}
- log.info("[{}] Resume the data replication after the schema fetching done", replicatorId);
- doRewindCursor(true);
});
} else {
msg.setSchemaInfoForReplicator(schemaFuture.get());
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 30538b119d88..87c467fbf182 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.Terminat
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.netty.buffer.ByteBuf;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
@@ -453,12 +454,16 @@ public abstract class PersistentReplicator extends AbstractReplicator
protected abstract boolean replicateEntries(List<Entry> entries, InFlightTask inFlightTask);
- protected CompletableFuture<SchemaInfo> getSchemaInfo(MessageImpl msg) throws ExecutionException {
- if (msg.getSchemaVersion() == null || msg.getSchemaVersion().length == 0) {
- return CompletableFuture.completedFuture(null);
+ protected CompletableFuture<SchemaInfo> 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() {
@@ -993,6 +998,26 @@ public abstract class PersistentReplicator extends AbstractReplicator
}
}
+ /**
+ * Rewinds the cursor like {@link #doRewindCursor(boolean)}, but keeps reads paused until the read failure
+ * backoff elapses, so a persistent failure, such as a failing schema lookup, does not turn into a busy retry
+ * loop. The rewinding reference taken by {@link #beforeTerminateOrCursorRewinding} is released by the retry.
+ */
+ protected void doRewindCursorAndDelayReadRetry() {
+ long waitTimeMillis;
+ synchronized (inFlightTasks) {
+ cursor.rewind();
+ waitTimeMillis = readFailureBackoff.next();
+ }
+ brokerService.executor().schedule(() -> {
+ synchronized (inFlightTasks) {
+ waitForCursorRewindingRefCnf -= 1;
+ reasonOfWaitForCursorRewinding = null;
+ }
+ readMoreEntries();
+ }, waitTimeMillis, TimeUnit.MILLISECONDS);
+ }
+
protected void doRewindCursor(boolean triggerReadMoreEntries) {
synchronized (inFlightTasks) {
cursor.rewind();
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 000000000000..a018a9b742fe
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/GeoPersistentReplicatorTest.java
@@ -0,0 +1,275 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThatThrownBy;
+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;
+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.ArrayList;
+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.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.api.schema.SchemaInfoProvider;
+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.mockito.ArgumentCaptor;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-replication")
+public class GeoPersistentReplicatorTest {
+
+ @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 testSchemaInfoSynchronousFailureReturnsFailedFuture(Throwable schemaLookupFailure) throws Exception {
+ ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(schemaLookupFailure);
+ MessageImpl<?> message = mock(MessageImpl.class);
+ when(message.getSchemaVersion()).thenReturn(new byte[] { 1 });
+
+ CompletableFuture<SchemaInfo> schemaFuture = replicator.getSchemaInfo(message);
+
+ assertThat(schemaFuture).isCompletedExceptionally();
+ assertThatThrownBy(schemaFuture::join).hasCause(schemaLookupFailure);
+ }
+
+ @Test(dataProvider = "synchronousSchemaLookupFailures")
+ public void testSchemaInfoSynchronousFailureDoesNotReadUntilScheduledRetry(
+ Throwable schemaLookupFailure) throws Exception {
+ ThrowingSchemaReplicator replicator = new ThrowingSchemaReplicator(schemaLookupFailure);
+ replicator.forceStarted();
+
+ Position firstPosition = PositionFactory.create(1, 2);
+ ByteBuf headersAndPayload = newMessageWithSchemaVersion();
+ Entry firstEntry = newEntry(firstPosition, headersAndPayload);
+ Entry secondEntry = mock(Entry.class);
+
+ List<Entry> entries = List.of(firstEntry, secondEntry);
+ InFlightTask inFlightTask = replicator.createOrRecycleInFlightTaskIntoQueue(firstPosition, entries.size());
+
+ try {
+ replicator.readEntriesComplete(entries, inFlightTask);
+
+ assertThat(inFlightTask.getCompletedEntries())
+ .as("the failed entry and skipped remaining entries should release their in-flight permits")
+ .isEqualTo(entries.size());
+ verify(firstEntry).release();
+ verify(secondEntry).release();
+ assertThat(headersAndPayload.refCnt())
+ .as("the discarded entry buffer should be released")
+ .isZero();
+ verify(replicator.cursor).rewind();
+ verify(replicator.cursor, never()).asyncReadEntriesOrWait(
+ anyInt(), anyLong(), any(), any(), any());
+ ArgumentCaptor<Long> retryDelay = ArgumentCaptor.forClass(Long.class);
+ // The paused readMoreEntries() call may also schedule its regular idle poll; the backoff retry is the
+ // first scheduled task.
+ verify(replicator.context.executor, atLeastOnce()).schedule(any(Runnable.class),
+ retryDelay.capture(), eq(TimeUnit.MILLISECONDS));
+ assertThat(retryDelay.getAllValues().get(0)).isPositive();
+
+ assertThat(replicator.context.scheduledTasks).isNotEmpty();
+ Runnable scheduledRetry = replicator.context.scheduledTasks.get(0);
+
+ scheduledRetry.run();
+ verify(replicator.cursor).rewind();
+ verify(replicator.cursor).asyncReadEntriesOrWait(
+ anyInt(), anyLong(), any(), any(), any());
+ } finally {
+ while (headersAndPayload.refCnt() > 0) {
+ headersAndPayload.release();
+ }
+ }
+ }
+
+ 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")
+ .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 {
+
+ private final ReplicatorContext context;
+
+ ThrowingSchemaReplicator(Throwable schemaLookupFailure)
+ throws PulsarServerException, ExecutionException {
+ this(new ReplicatorContext(schemaLookupFailure));
+ }
+
+ private ThrowingSchemaReplicator(ReplicatorContext context)
+ throws PulsarServerException {
+ super(context.topic, context.cursor, "local", "remote", context.brokerService,
+ context.replicationClient);
+ this.context = context;
+ }
+
+ @Override
+ protected void startProducer() {
+ // Avoid creating a real remote producer from the superclass constructor.
+ }
+
+ void forceStarted() {
+ STATE_UPDATER.set(this, State.Started);
+ this.producer = mock(ProducerImpl.class);
+ }
+
+ private static PersistentTopic mockTopic(BrokerService brokerService) {
+ PersistentTopic topic = mock(PersistentTopic.class);
+ 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");
+ when(cursor.getReadPosition()).thenReturn(PositionFactory.create(1, 1));
+ return cursor;
+ }
+
+ private static BrokerService mockBrokerService(EventLoopGroup executor,
+ List<Runnable> scheduledTasks, Throwable schemaLookupFailure)
+ throws PulsarServerException, ExecutionException {
+ ServiceConfiguration config = new ServiceConfiguration();
+ PulsarService pulsar = mock(PulsarService.class);
+ BrokerService brokerService = mock(BrokerService.class);
+ PulsarClientImpl localClient = mockLocalClient(schemaLookupFailure);
+ 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);
+ when(brokerService.executor()).thenReturn(executor);
+ doAnswer(invocation -> {
+ scheduledTasks.add(invocation.getArgument(0, Runnable.class));
+ return null;
+ }).when(executor).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
+ return brokerService;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PulsarClientImpl mockLocalClient(Throwable schemaLookupFailure)
+ throws ExecutionException {
+ PulsarClientImpl localClient = mock(PulsarClientImpl.class);
+ LoadingCache<String, SchemaInfoProvider> schemaProviderLoadingCache = mock(LoadingCache.class);
+ when(localClient.getSchemaProviderLoadingCache()).thenReturn(schemaProviderLoadingCache);
+ doAnswer(__ -> {
+ throw schemaLookupFailure;
+ }).when(schemaProviderLoadingCache).get(anyString());
+ return localClient;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static PulsarClientImpl mockReplicationClient() {
+ PulsarClientImpl replicationClient = mock(PulsarClientImpl.class);
+ ProducerBuilder<byte[]> producerBuilder = mock(ProducerBuilder.class, RETURNS_SELF);
+ when(replicationClient.newProducer(any(Schema.class))).thenReturn(producerBuilder);
+ return replicationClient;
+ }
+ }
+
+ private static class ReplicatorContext {
+ private final List<Runnable> scheduledTasks;
+ private final EventLoopGroup executor;
+ private final BrokerService brokerService;
+ private final PersistentTopic topic;
+ private final ManagedCursor cursor;
+ private final PulsarClientImpl replicationClient;
+
+ private ReplicatorContext(Throwable schemaLookupFailure)
+ throws PulsarServerException, ExecutionException {
+ this.scheduledTasks = new ArrayList<>();
+ this.executor = mock(EventLoopGroup.class);
+ this.brokerService = ThrowingSchemaReplicator.mockBrokerService(executor, scheduledTasks,
+ schemaLookupFailure);
+ this.topic = ThrowingSchemaReplicator.mockTopic(brokerService);
+ this.cursor = ThrowingSchemaReplicator.mockCursor();
+ this.replicationClient = ThrowingSchemaReplicator.mockReplicationClient();
+ }
+ }
+}