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/02/06 12:37:39 UTC
(pulsar) branch branch-4.1 updated (eaf42b2569c -> 13a8f678ee6)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from eaf42b2569c [fix][broker] Prevent missed topic changes in topic watchers and schedule periodic refresh with patternAutoDiscoveryPeriod interval (#25188)
new 05157e71d63 [fix][sec] Upgrade OpenSearch to 2.19.4 to remediate CVE-2025-9624 (#25206)
new 8f791f0315d [improve][admin] Add client side looping to analyze-backlog in Topics to avoid potential HTTP call timeout (#25127)
new 2057fbd5887 [feat][io] implement pip-297 for jdbc sinks (#25195)
new 7d44375b20f [improve][broker] Change log level from warn to debug when cursor mark-deleted position ledger doesn't exist (#25200)
new 13a8f678ee6 [fix][broker] Fix httpProxyTimeout config (#25223)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 3 +-
.../org/apache/bookkeeper/mledger/impl/OpScan.java | 34 ++--
pom.xml | 2 +-
.../admin/AnalyzeBacklogSubscriptionTest.java | 195 +++++++++++++++++++++
.../org/apache/pulsar/client/admin/Topics.java | 116 +++++++++++-
.../pulsar/client/admin/internal/TopicsImpl.java | 112 ++++++++++++
.../apache/pulsar/io/jdbc/JdbcAbstractSink.java | 25 ++-
.../apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java | 74 +++++++-
.../pulsar/proxy/server/AdminProxyHandler.java | 10 +-
.../pulsar/proxy/server/AdminProxyHandlerTest.java | 22 ++-
10 files changed, 562 insertions(+), 31 deletions(-)
(pulsar) 05/05: [fix][broker] Fix httpProxyTimeout config (#25223)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 13a8f678ee6023b50df54a57b12085581b58f319
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Fri Feb 6 20:28:25 2026 +0800
[fix][broker] Fix httpProxyTimeout config (#25223)
Co-authored-by: Copilot <17...@users.noreply.github.com>
(cherry picked from commit 2d6ef6f288b7a105c59c176f59d5b3705d106e1b)
---
.../pulsar/proxy/server/AdminProxyHandler.java | 10 +++++++---
.../pulsar/proxy/server/AdminProxyHandlerTest.java | 22 +++++++++++++++++++---
2 files changed, 26 insertions(+), 6 deletions(-)
diff --git a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
index 08f87f3d08a..090f2f4a384 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
@@ -109,7 +109,6 @@ class AdminProxyHandler extends ProxyServlet {
TimeUnit.SECONDS);
}
}
- super.setTimeout(config.getHttpProxyTimeout());
}
@Override
@@ -119,10 +118,15 @@ class AdminProxyHandler extends ProxyServlet {
return httpClient;
}
- private void customizeHttpClient(HttpClient httpClient) {
+ protected void customizeHttpClient(HttpClient httpClient) {
httpClient.setFollowRedirects(true);
+
ProtocolHandlers protocolHandlers = httpClient.getProtocolHandlers();
- protocolHandlers.put(new RedirectProtocolHandler(httpClient));
+ if (protocolHandlers != null) {
+ protocolHandlers.put(new RedirectProtocolHandler(httpClient));
+ }
+
+ setTimeout(config.getHttpProxyTimeout());
}
// This class allows the request body to be replayed, the default implementation
diff --git a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AdminProxyHandlerTest.java b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AdminProxyHandlerTest.java
index fdf9242c9f3..151ffc79f0b 100644
--- a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AdminProxyHandlerTest.java
+++ b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AdminProxyHandlerTest.java
@@ -20,7 +20,9 @@ package org.apache.pulsar.proxy.server;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
@@ -59,6 +61,20 @@ public class AdminProxyHandlerTest {
adminProxyHandler.init(servletConfig);
}
+ @Test
+ public void testRequestTimeout() {
+ ProxyConfiguration proxyConfiguration = spy(new ProxyConfiguration());
+ proxyConfiguration.setHttpProxyTimeout(120 * 1000);
+
+ adminProxyHandler = new AdminProxyHandler(proxyConfiguration,
+ mock(BrokerDiscoveryProvider.class), mock(Authentication.class));
+
+ HttpClient httpClient = mock(HttpClient.class);
+ adminProxyHandler.customizeHttpClient(httpClient);
+
+ assertEquals(adminProxyHandler.getTimeout(), 120 * 1000);
+ }
+
@Test
public void replayableProxyContentProviderTest() throws Exception {
HttpServletRequest request = mock(HttpServletRequest.class);
@@ -102,10 +118,10 @@ public class AdminProxyHandlerTest {
}
// then
- Assert.assertEquals(consumedBytes, requestBodySize);
+ assertEquals(consumedBytes, requestBodySize);
Field field = replayableProxyContentProvider.getClass().getDeclaredField("bodyBufferMaxSizeReached");
field.setAccessible(true);
- Assert.assertEquals(((boolean) field.get(replayableProxyContentProvider)), true);
+ assertEquals(((boolean) field.get(replayableProxyContentProvider)), true);
}
@Test
@@ -138,7 +154,7 @@ public class AdminProxyHandlerTest {
byte[] consumedBytes = new byte[consumeBuffer.limit()];
consumeBuffer.get(consumedBytes);
// then
- Assert.assertEquals(consumedBytes, inputBuffer);
+ assertEquals(consumedBytes, inputBuffer, "i=" + i);
}
}
}
(pulsar) 03/05: [feat][io] implement pip-297 for jdbc sinks (#25195)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 2057fbd5887bfff16c8f26009c9499fb4f6a1b80
Author: Bäm <de...@sandchaschte.ch>
AuthorDate: Fri Feb 6 12:15:52 2026 +0100
[feat][io] implement pip-297 for jdbc sinks (#25195)
(cherry picked from commit 6f4ac21eeed7bf9a8d4306c17468ce55b74f84a1)
---
.../apache/pulsar/io/jdbc/JdbcAbstractSink.java | 25 +++++++-
.../apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java | 74 +++++++++++++++++++++-
2 files changed, 97 insertions(+), 2 deletions(-)
diff --git a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
index ca33b3cfdab..73ba6b712f0 100644
--- a/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
+++ b/pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/JdbcAbstractSink.java
@@ -35,6 +35,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import lombok.AllArgsConstructor;
@@ -50,6 +51,11 @@ import org.apache.pulsar.io.core.SinkContext;
*/
@Slf4j
public abstract class JdbcAbstractSink<T> implements Sink<T> {
+
+ private enum State {
+ OPEN, FAILED, CLOSED
+ }
+
// ----- Runtime fields
protected JdbcSinkConfig jdbcSinkConfig;
@Getter
@@ -73,9 +79,12 @@ public abstract class JdbcAbstractSink<T> implements Sink<T> {
private AtomicBoolean isFlushing;
private int batchSize;
private ScheduledExecutorService flushExecutor;
+ private SinkContext sinkContext;
+ private final AtomicReference<State> state = new AtomicReference<>(State.OPEN);
@Override
public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
+ this.sinkContext = sinkContext;
jdbcSinkConfig = JdbcSinkConfig.load(config, sinkContext);
jdbcSinkConfig.validate();
@@ -148,6 +157,7 @@ public abstract class JdbcAbstractSink<T> implements Sink<T> {
@Override
public void close() throws Exception {
+ state.set(State.CLOSED);
if (flushExecutor != null) {
int timeoutMs = jdbcSinkConfig.getTimeoutMs() * 2;
flushExecutor.shutdown();
@@ -310,8 +320,9 @@ public abstract class JdbcAbstractSink<T> implements Sink<T> {
connection.rollback();
}
} catch (Exception ex) {
- throw new RuntimeException(ex);
+ log.error("Failed to rollback transaction", ex);
}
+ fatal(e);
}
isFlushing.set(false);
@@ -385,4 +396,16 @@ public abstract class JdbcAbstractSink<T> implements Sink<T> {
return true;
}
+ /**
+ * Signal a fatal exception to the framework.
+ * This will cause the function instance to terminate properly.
+ *
+ * @param e the fatal exception
+ */
+ private void fatal(Exception e) {
+ if (sinkContext != null && state.compareAndSet(State.OPEN, State.FAILED)) {
+ sinkContext.fatal(e);
+ }
+ }
+
}
diff --git a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java
index 901ac9f1e39..7cab33df309 100644
--- a/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java
+++ b/pulsar-io/jdbc/sqlite/src/test/java/org/apache/pulsar/io/jdbc/SqliteJdbcSinkTest.java
@@ -18,11 +18,16 @@
*/
package org.apache.pulsar.io.jdbc;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -56,6 +61,7 @@ import org.apache.pulsar.common.schema.KeyValueEncodingType;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.functions.source.PulsarRecord;
+import org.apache.pulsar.io.core.SinkContext;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
@@ -133,7 +139,9 @@ public class SqliteJdbcSinkTest {
@AfterMethod(alwaysRun = true)
public void tearDown() throws Exception {
- jdbcSink.close();
+ if (jdbcSink != null) {
+ jdbcSink.close();
+ }
sqliteUtils.tearDown();
}
@@ -860,6 +868,70 @@ public class SqliteJdbcSinkTest {
}
}
+ /**
+ * Test that fatal() is called when an unrecoverable exception occurs during flush.
+ * This verifies the PIP-297 implementation for proper termination of the sink.
+ *
+ * The test works by:
+ * 1. Opening the sink with a valid table (so open() succeeds)
+ * 2. Using reflection to replace the insertStatement with a mock that throws SQLException
+ * 3. Writing a record to trigger flush
+ * 4. Verifying that fatal() was called with the exception
+ */
+ @Test
+ public void testFatalCalledOnFlushException() throws Exception {
+ jdbcSink.close();
+ jdbcSink = null;
+
+ String jdbcUrl = sqliteUtils.sqliteUri();
+ Map<String, Object> conf = Maps.newHashMap();
+ conf.put("jdbcUrl", jdbcUrl);
+ conf.put("tableName", tableName); // Use valid table so open() succeeds
+ conf.put("key", "field3");
+ conf.put("nonKey", "field1,field2");
+ conf.put("batchSize", 1);
+
+ SinkContext mockSinkContext = mock(SinkContext.class);
+ AtomicReference<Throwable> fatalException = new AtomicReference<>();
+ doAnswer(invocation -> {
+ fatalException.set(invocation.getArgument(0));
+ return null;
+ }).when(mockSinkContext).fatal(any(Throwable.class));
+
+ SqliteJdbcAutoSchemaSink sinkWithContext = new SqliteJdbcAutoSchemaSink();
+ try {
+ sinkWithContext.open(conf, mockSinkContext);
+
+ // Create a mock PreparedStatement that throws SQLException on execute()
+ PreparedStatement mockStatement = mock(PreparedStatement.class);
+ SQLException simulatedException = new SQLException("Simulated database connection failure");
+ doThrow(simulatedException).when(mockStatement).execute();
+ doThrow(simulatedException).when(mockStatement).executeBatch();
+
+ // Use reflection to replace the insertStatement with our mock
+ FieldUtils.writeField(sinkWithContext, "insertStatement", mockStatement, true);
+
+ Foo insertObj = new Foo("f1", "f2", 1);
+ Map<String, String> props = Maps.newHashMap();
+ props.put("ACTION", "INSERT");
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ sinkWithContext.write(createMockFooRecord(insertObj, props, future));
+
+ // Wait for the flush to complete and fail
+ Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
+ verify(mockSinkContext).fatal(any(Throwable.class));
+ Assert.assertNotNull(fatalException.get());
+ Assert.assertTrue(fatalException.get() instanceof SQLException);
+ Assert.assertEquals(fatalException.get().getMessage(), "Simulated database connection failure");
+ });
+
+ // Verify the record was failed (not acked)
+ Assert.assertFalse(future.get(1, TimeUnit.SECONDS));
+ } finally {
+ sinkWithContext.close();
+ }
+ }
+
@SuppressWarnings("unchecked")
private Record<GenericObject> createMockFooRecord(Foo record, Map<String, String> actionProperties,
CompletableFuture<Boolean> future) {
(pulsar) 02/05: [improve][admin] Add client side looping to analyze-backlog in Topics to avoid potential HTTP call timeout (#25127)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 8f791f0315d7a20e8984cf13ee8487ea2525619e
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Fri Feb 6 18:05:42 2026 +0800
[improve][admin] Add client side looping to analyze-backlog in Topics to avoid potential HTTP call timeout (#25127)
(cherry picked from commit e160b1add39945148150365aefcabbc77f13b759)
---
.../org/apache/bookkeeper/mledger/impl/OpScan.java | 34 ++--
.../admin/AnalyzeBacklogSubscriptionTest.java | 195 +++++++++++++++++++++
.../org/apache/pulsar/client/admin/Topics.java | 116 +++++++++++-
.../pulsar/client/admin/internal/TopicsImpl.java | 112 ++++++++++++
4 files changed, 436 insertions(+), 21 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java
index 732071ee01a..413bb5c018e 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpScan.java
@@ -69,29 +69,23 @@ class OpScan implements ReadEntriesCallback {
try {
Position lastPositionForBatch = entries.get(entries.size() - 1).getPosition();
lastSeenPosition = lastPositionForBatch;
- // filter out the entry if it has been already deleted
- // filterReadEntries will call entry.release if the entry is filtered out
- List<Entry> entriesFiltered = this.cursor.filterReadEntries(entries);
- int skippedEntries = entries.size() - entriesFiltered.size();
- remainingEntries.addAndGet(-skippedEntries);
- if (!entriesFiltered.isEmpty()) {
- for (Entry entry : entriesFiltered) {
- if (remainingEntries.decrementAndGet() <= 0) {
- log.warn("[{}] Scan abort after reading too many entries", OpScan.this.cursor);
- callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);
- return;
- }
- if (!condition.test(entry)) {
- log.warn("[{}] Scan abort due to user code", OpScan.this.cursor);
- callback.scanComplete(lastSeenPosition, ScanOutcome.USER_INTERRUPTED, OpScan.this.ctx);
- return;
- }
+ for (Entry entry : entries) {
+ if (remainingEntries.getAndDecrement() <= 0) {
+ log.info("[{}] Scan abort after reading too many entries", OpScan.this.cursor);
+ callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);
+ return;
+ }
+ if (!condition.test(entry)) {
+ log.info("[{}] Scan abort due to user code", OpScan.this.cursor);
+ callback.scanComplete(lastSeenPosition, ScanOutcome.USER_INTERRUPTED, OpScan.this.ctx);
+ return;
}
}
searchPosition = ledger.getPositionAfterN(lastPositionForBatch, 1,
PositionBound.startExcluded);
if (log.isDebugEnabled()) {
- log.debug("readEntryComplete {} at {} next is {}", lastPositionForBatch, searchPosition);
+ log.debug("[{}] readEntryComplete at {} next is {}", OpScan.this.cursor, lastPositionForBatch,
+ searchPosition);
}
if (searchPosition.compareTo(lastPositionForBatch) == 0) {
@@ -117,12 +111,12 @@ class OpScan implements ReadEntriesCallback {
public void find() {
if (remainingEntries.get() <= 0) {
- log.warn("[{}] Scan abort after reading too many entries", OpScan.this.cursor);
+ log.info("[{}] Scan abort after reading too many entries", OpScan.this.cursor);
callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);
return;
}
if (System.currentTimeMillis() - startTime > timeOutMs) {
- log.warn("[{}] Scan abort after hitting the deadline", OpScan.this.cursor);
+ log.info("[{}] Scan abort after hitting the deadline", OpScan.this.cursor);
callback.scanComplete(lastSeenPosition, ScanOutcome.ABORTED, OpScan.this.ctx);
return;
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AnalyzeBacklogSubscriptionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AnalyzeBacklogSubscriptionTest.java
index acea9132049..4425436954a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AnalyzeBacklogSubscriptionTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AnalyzeBacklogSubscriptionTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.admin;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertThrows;
import java.util.ArrayList;
import java.util.List;
@@ -27,10 +28,12 @@ import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
+import org.apache.commons.lang3.RandomUtils;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.SubscriptionType;
@@ -50,6 +53,12 @@ public class AnalyzeBacklogSubscriptionTest extends ProducerConsumerBase {
producerBaseSetup();
}
+ @Override
+ protected void doInitConf() throws Exception {
+ super.doInitConf();
+ conf.setDispatcherMaxReadBatchSize(10);
+ }
+
@AfterMethod(alwaysRun = true)
@Override
public void cleanup() throws Exception {
@@ -189,4 +198,190 @@ public class AnalyzeBacklogSubscriptionTest extends ProducerConsumerBase {
assertEquals(0, analyzeSubscriptionBacklogResult.getEntries());
}
+ @Test
+ public void analyzeBacklogServerReturnFalseAbortedFlagWithoutLoop() throws Exception {
+ int serverSubscriptionBacklogScanMaxEntries = 20;
+ conf.setSubscriptionBacklogScanMaxEntries(serverSubscriptionBacklogScanMaxEntries);
+
+ String topic = "persistent://my-property/my-ns/analyze-backlog-server-return-false-aborted-flag-without-loop";
+ String subName = "sub-1";
+ int numMessages = 10;
+
+ // Test server returns false aborted flag.
+ List<MessageId> messageIds = clientSideLoopAnalyzeBacklogSetup(topic, subName, numMessages);
+
+ verifyClientSideLoopBacklog(topic, subName, numMessages - 1, numMessages, messageIds.get(0),
+ messageIds.get(numMessages - 1));
+ }
+
+ @Test
+ public void analyzeBacklogMaxEntriesExceedWithoutLoop() throws Exception {
+ int serverSubscriptionBacklogScanMaxEntries = 20;
+ conf.setSubscriptionBacklogScanMaxEntries(serverSubscriptionBacklogScanMaxEntries);
+
+ String topic = "persistent://my-property/my-ns/analyze-backlog-max-entries-exceed-without-loop";
+ String subName = "sub-1";
+ int numMessages = 25;
+
+ // Test backlogScanMaxEntries(client side) <= subscriptionBacklogScanMaxEntries(server side), but server
+ // returns true aborted flag. Server dispatcherMaxReadBatchSize is set to 10.
+ List<MessageId> messageIds = clientSideLoopAnalyzeBacklogSetup(topic, subName, numMessages);
+
+ verifyClientSideLoopBacklog(topic, subName, serverSubscriptionBacklogScanMaxEntries - 1,
+ serverSubscriptionBacklogScanMaxEntries, messageIds.get(0),
+ messageIds.get(serverSubscriptionBacklogScanMaxEntries - 1));
+
+ }
+
+ @Test
+ public void analyzeBacklogServerReturnFalseAbortedFlagWithLoop() throws Exception {
+ int serverSubscriptionBacklogScanMaxEntries = 20;
+ conf.setSubscriptionBacklogScanMaxEntries(serverSubscriptionBacklogScanMaxEntries);
+
+ String topic = "persistent://my-property/my-ns/analyze-backlog-server-return-false-aborted-flag-with-loop";
+ String subName = "sub-1";
+ int numMessages = 45;
+
+ // Test client side loop: backlogScanMaxEntries > subscriptionBacklogScanMaxEntries, the loop termination
+ // condition is that server returns false aborted flag.
+ List<MessageId> messageIds = clientSideLoopAnalyzeBacklogSetup(topic, subName, numMessages);
+
+ verifyClientSideLoopBacklog(topic, subName, numMessages, numMessages, messageIds.get(0),
+ messageIds.get(numMessages - 1));
+ }
+
+ @Test
+ public void analyzeBacklogMaxEntriesExceedWithLoop() throws Exception {
+ int serverSubscriptionBacklogScanMaxEntries = 15;
+ conf.setSubscriptionBacklogScanMaxEntries(serverSubscriptionBacklogScanMaxEntries);
+
+ String topic = "persistent://my-property/my-ns/analyze-backlog-max-entries-exceed-with-loop";
+ String subName = "sub-1";
+ int numMessages = 55;
+ int backlogScanMaxEntries = 40;
+
+ // Test client side loop: backlogScanMaxEntries > subscriptionBacklogScanMaxEntries, the loop termination
+ // condition is that total entries exceeds backlogScanMaxEntries.
+ // Server dispatcherMaxReadBatchSize is set to 10.
+ List<MessageId> messageIds = clientSideLoopAnalyzeBacklogSetup(topic, subName, numMessages);
+
+ // Broker returns 15 + 15 + 15 = 45 entries.
+ int expectedEntries = (backlogScanMaxEntries / serverSubscriptionBacklogScanMaxEntries + 1)
+ * serverSubscriptionBacklogScanMaxEntries;
+ verifyClientSideLoopBacklog(topic, subName, backlogScanMaxEntries, expectedEntries, messageIds.get(0),
+ messageIds.get(expectedEntries - 1));
+ }
+
+ @Test
+ public void analyzeBacklogWithTopicUnload() throws Exception {
+ int serverSubscriptionBacklogScanMaxEntries = 10;
+ conf.setSubscriptionBacklogScanMaxEntries(serverSubscriptionBacklogScanMaxEntries);
+
+ String topic = "persistent://my-property/my-ns/analyze-backlog-with-topic-unload";
+ String subName = "sub-1";
+ int numMessages = 35;
+
+ admin.topics().createSubscription(topic, subName, MessageId.latest);
+
+ assertEquals(admin.topics().getSubscriptions(topic), List.of("sub-1"));
+ verifyBacklog(topic, subName, 0, 0);
+
+ // Test client side loop with topic unload. Use sync send method here to avoid potential message duplication.
+ @Cleanup Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).enableBatching(false).create();
+ List<MessageId> messageIds = new ArrayList<>();
+ for (int i = 0; i < numMessages; i++) {
+ MessageId messageId = producer.send(("test-" + i).getBytes());
+ messageIds.add(messageId);
+ if (RandomUtils.secure().randomBoolean()) {
+ admin.topics().unload(topic);
+ }
+ }
+
+ verifyClientSideLoopBacklog(topic, subName, numMessages, numMessages, messageIds.get(0),
+ messageIds.get(numMessages - 1));
+ }
+
+ @Test
+ public void analyzeBacklogWithIndividualAck() throws Exception {
+ int serverSubscriptionBacklogScanMaxEntries = 20;
+ conf.setSubscriptionBacklogScanMaxEntries(serverSubscriptionBacklogScanMaxEntries);
+
+ String topic = "persistent://my-property/my-ns/analyze-backlog-with-individual-ack";
+ String subName = "sub-1";
+ int messages = 55;
+
+ // Test client side loop with individual ack.
+ List<MessageId> messageIds = clientSideLoopAnalyzeBacklogSetup(topic, subName, messages);
+
+ // We want to wait for the server to process acks, in order to not have a flaky test.
+ @Cleanup Consumer<byte[]> consumer =
+ pulsarClient.newConsumer().topic(topic).isAckReceiptEnabled(true).subscriptionName(subName)
+ .subscriptionType(SubscriptionType.Shared).subscribe();
+
+ // Individual ack message2.
+ Message<byte[]> message1 = consumer.receive();
+ Message<byte[]> message2 = consumer.receive();
+ consumer.acknowledge(message2);
+
+ int backlogScanMaxEntries = 20;
+ verifyClientSideLoopBacklog(topic, subName, backlogScanMaxEntries, backlogScanMaxEntries, messageIds.get(0),
+ messageIds.get(backlogScanMaxEntries));
+
+ // Ack message1.
+ consumer.acknowledge(message1);
+ verifyClientSideLoopBacklog(topic, subName, backlogScanMaxEntries, backlogScanMaxEntries, messageIds.get(2),
+ messageIds.get(backlogScanMaxEntries + 1));
+
+ // Ack all messages.
+ for (int i = 2; i < messages; i++) {
+ Message<byte[]> message = consumer.receive();
+ consumer.acknowledge(message);
+ }
+
+ verifyClientSideLoopBacklog(topic, subName, backlogScanMaxEntries, 0, null, null);
+ }
+
+ private List<MessageId> clientSideLoopAnalyzeBacklogSetup(String topic, String subName, int numMessages)
+ throws Exception {
+ admin.topics().createSubscription(topic, subName, MessageId.latest);
+
+ assertEquals(admin.topics().getSubscriptions(topic), List.of("sub-1"));
+ verifyClientSideLoopBacklog(topic, subName, -1, 0, null, null);
+
+ @Cleanup Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).enableBatching(false).create();
+ List<CompletableFuture<MessageId>> futures = new ArrayList<>();
+ for (int i = 0; i < numMessages; i++) {
+ CompletableFuture<MessageId> future = producer.sendAsync(("test-" + i).getBytes());
+ futures.add(future);
+ }
+ FutureUtil.waitForAll(futures).get();
+ return futures.stream().map(CompletableFuture::join).toList();
+ }
+
+ private void verifyClientSideLoopBacklog(String topic, String subName, int backlogMaxScanEntries,
+ int expectedEntries, MessageId firstMessageId, MessageId lastMessageId)
+ throws Exception {
+ AnalyzeSubscriptionBacklogResult backlogResult =
+ admin.topics().analyzeSubscriptionBacklog(topic, subName, Optional.empty(), backlogMaxScanEntries);
+
+ assertEquals(backlogResult.getEntries(), expectedEntries);
+ assertEquals(backlogResult.getMessages(), expectedEntries);
+
+ if (firstMessageId == null) {
+ assertNull(backlogResult.getFirstMessageId());
+ } else {
+ MessageIdAdv firstMessageIdAdv = (MessageIdAdv) firstMessageId;
+ assertEquals(backlogResult.getFirstMessageId(),
+ firstMessageIdAdv.getLedgerId() + ":" + firstMessageIdAdv.getEntryId());
+ }
+
+ if (lastMessageId == null) {
+ assertNull(backlogResult.getLastMessageId());
+ } else {
+ MessageIdAdv lastMessageIdAdv = (MessageIdAdv) lastMessageId;
+ assertEquals(backlogResult.getLastMessageId(),
+ lastMessageIdAdv.getLedgerId() + ":" + lastMessageIdAdv.getEntryId());
+ }
+ }
+
}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
index a2fcd60deb5..4360d37433f 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Topics.java
@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException;
import org.apache.pulsar.client.admin.PulsarAdminException.NotAllowedException;
import org.apache.pulsar.client.admin.PulsarAdminException.NotAuthorizedException;
@@ -2220,21 +2221,134 @@ public interface Topics {
* This is a potentially expensive operation, as it requires
* to read the messages from storage.
* This function takes into consideration batch messages
- * and also Subscription filters.
+ * and also Subscription filters. <br/>
+ * See also: {@link #analyzeSubscriptionBacklogAsync(String, String, Optional, long)}
* @param topic
* Topic name
* @param subscriptionName
* the subscription
* @param startPosition
* the position to start the scan from (empty means the last processed message)
+ * @param backlogScanMaxEntries
+ * the maximum number of backlog entries the client will scan before terminating its loop
* @return an accurate analysis of the backlog
* @throws PulsarAdminException
* Unexpected error
*/
+ AnalyzeSubscriptionBacklogResult analyzeSubscriptionBacklog(String topic, String subscriptionName,
+ Optional<MessageId> startPosition,
+ long backlogScanMaxEntries) throws PulsarAdminException;
+
+ /**
+ * Analyze subscription backlog.
+ * This is a potentially expensive operation, as it requires
+ * to read the messages from storage.
+ * This function takes into consideration batch messages
+ * and also Subscription filters. <br/>
+ * See also: {@link #analyzeSubscriptionBacklogAsync(String, String, Optional, Predicate)} <br/>
+ *
+ * @param topic
+ * Topic name
+ * @param subscriptionName
+ * the subscription
+ * @param startPosition
+ * the position to start the scan from (empty means the last processed message)
+ * @param terminatePredicate
+ * the predicate to determine whether to terminate the loop
+ * @return an accurate analysis of the backlog
+ */
+ AnalyzeSubscriptionBacklogResult analyzeSubscriptionBacklog(String topic, String subscriptionName,
+ Optional<MessageId> startPosition,
+ Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate)
+ throws PulsarAdminException;
+
+ /**
+ * Analyze subscription backlog.
+ * This is a potentially expensive operation, as it requires
+ * to read the messages from storage.
+ * This function takes into consideration batch messages
+ * and also Subscription filters.
+ * @param topic
+ * Topic name
+ * @param subscriptionName
+ * the subscription
+ * @param startPosition
+ * the position to start the scan from (empty means the last processed message)
+ * @return an accurate analysis of the backlog
+ */
CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic,
String subscriptionName,
Optional<MessageId> startPosition);
+ /**
+ * Analyze subscription backlog.
+ * This is a potentially expensive operation, as it requires
+ * to read the messages from storage.
+ * This function takes into consideration batch messages
+ * and also Subscription filters.
+ *
+ *<p>
+ * What's the purpose of this overloaded method? <br/>
+ * There are broker side configurable maximum limits how many entries will be read and how long the scanning can
+ * take. The subscriptionBacklogScanMaxTimeMs (default 2 minutes) and subscriptionBacklogScanMaxEntries
+ * (default 10000) control this behavior. <br/>
+ * Increasing these settings is possible. However, it's possible that the HTTP request times out (also idle timeout
+ * in NAT/firewall etc.) before the command completes so increasing the limits might not be useful beyond a few
+ * minutes.
+ *</p>
+ *
+ *<p>
+ * How does this method work? <br/>
+ * 1. Add a new parameter backlogScanMaxEntries in client side method to control the client-side loop termination
+ * condition. <br/>
+ * 2. If subscriptionBacklogScanMaxEntries(server side) >= backlogScanMaxEntries(client side), then
+ * backlogScanMaxEntries parameter will take no effect. <br/>
+ * 3. If subscriptionBacklogScanMaxEntries < backlogScanMaxEntries, the client will call analyze-backlog method in
+ * a loop until server return ScanOutcome.COMPLETED or the total entries exceeds backlogScanMaxEntries. <br/>
+ * 4. This means that backlogScanMaxEntries cannot be used to precisely control the number of entries scanned by
+ * the server, it only serves to determine when the loop should terminate. <br/>
+ * 5. With this method, the server can reduce the values of the two parameters subscriptionBacklogScanMaxTimeMs and
+ * subscriptionBacklogScanMaxEntries, so user can retrieve the desired number of backlog entries through
+ * client-side looping.
+ *</p>
+ * @param topic
+ * Topic name
+ * @param subscriptionName
+ * the subscription
+ * @param startPosition
+ * the position to start the scan from (empty means the last processed message)
+ * @param backlogScanMaxEntries
+ * the maximum number of backlog entries the client will scan before terminating its loop
+ * @return an accurate analysis of the backlog
+ */
+ CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic,
+ String subscriptionName,
+ Optional<MessageId> startPosition,
+ long backlogScanMaxEntries);
+
+ /**
+ * Analyze subscription backlog.
+ * This is a potentially expensive operation, as it requires
+ * to read the messages from storage.
+ * This function takes into consideration batch messages
+ * and also Subscription filters. <br/>
+ * See also: {@link #analyzeSubscriptionBacklogAsync(String, String, Optional, long)} <br/>
+ * User can control the loop termination condition by terminatePredicate.
+ *
+ * @param topic
+ * Topic name
+ * @param subscriptionName
+ * the subscription
+ * @param startPosition
+ * the position to start the scan from (empty means the last processed message)
+ * @param terminatePredicate
+ * the predicate to determine whether to terminate the loop
+ * @return an accurate analysis of the backlog
+ */
+ CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic,
+ String subscriptionName, Optional<MessageId> startPosition,
+ Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate);
+
/**
* Get backlog size by a message ID.
* @param topic
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
index 11dd69a23ce..78b7329159c 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
@@ -34,6 +34,10 @@ import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiConsumer;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.ws.rs.client.Entity;
@@ -44,6 +48,7 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.admin.GetStatsOptions;
import org.apache.pulsar.client.admin.ListTopicsOptions;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
@@ -1560,6 +1565,23 @@ public class TopicsImpl extends BaseResource implements Topics {
return sync(() -> analyzeSubscriptionBacklogAsync(topic, subscriptionName, startPosition));
}
+ @Override
+ public AnalyzeSubscriptionBacklogResult analyzeSubscriptionBacklog(String topic, String subscriptionName,
+ Optional<MessageId> startPosition,
+ long backlogScanMaxEntries)
+ throws PulsarAdminException {
+ return sync(
+ () -> analyzeSubscriptionBacklogAsync(topic, subscriptionName, startPosition, backlogScanMaxEntries));
+ }
+
+ @Override
+ public AnalyzeSubscriptionBacklogResult analyzeSubscriptionBacklog(String topic, String subscriptionName,
+ Optional<MessageId> startPosition,
+ Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate)
+ throws PulsarAdminException {
+ return sync(() -> analyzeSubscriptionBacklogAsync(topic, subscriptionName, startPosition, terminatePredicate));
+ }
+
@Override
public CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic,
String subscriptionName,
@@ -1591,6 +1613,96 @@ public class TopicsImpl extends BaseResource implements Topics {
return future;
}
+ @Override
+ public CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic,
+ String subscriptionName,
+ Optional<MessageId> startPosition,
+ long backlogScanMaxEntries) {
+ return analyzeSubscriptionBacklogAsync(topic, subscriptionName, startPosition,
+ (backlogResult) -> backlogResult.getEntries() >= backlogScanMaxEntries);
+ }
+
+ @Override
+ public CompletableFuture<AnalyzeSubscriptionBacklogResult> analyzeSubscriptionBacklogAsync(String topic,
+ String subscriptionName, Optional<MessageId> startPosition,
+ Predicate<AnalyzeSubscriptionBacklogResult> terminatePredicate) {
+ final CompletableFuture<AnalyzeSubscriptionBacklogResult> future = new CompletableFuture<>();
+ AtomicReference<AnalyzeSubscriptionBacklogResult> resultRef = new AtomicReference<>();
+ int partitionIndex = TopicName.get(topic).getPartitionIndex();
+ AtomicReference<Optional<MessageId>> startPositionRef = new AtomicReference<>(startPosition);
+
+ Supplier<CompletableFuture<AnalyzeSubscriptionBacklogResult>> resultSupplier =
+ () -> analyzeSubscriptionBacklogAsync(topic, subscriptionName, startPositionRef.get());
+ BiConsumer<AnalyzeSubscriptionBacklogResult, Throwable> completeAction = new BiConsumer<>() {
+ @Override
+ public void accept(AnalyzeSubscriptionBacklogResult currentResult, Throwable throwable) {
+ if (throwable != null) {
+ future.completeExceptionally(throwable);
+ return;
+ }
+
+ AnalyzeSubscriptionBacklogResult mergedResult = mergeBacklogResults(currentResult, resultRef.get());
+ resultRef.set(mergedResult);
+ if (!mergedResult.isAborted() || terminatePredicate.test(mergedResult)) {
+ future.complete(mergedResult);
+ return;
+ }
+
+ // In analyze-backlog, we treat 0 entries or null lastMessageId as scan completed for mere safety.
+ // 0 entries or a null lastMessageId indicates no entries were scanned.
+ if (currentResult.getEntries() <= 0 || StringUtils.isBlank(currentResult.getLastMessageId())) {
+ log.info("[{}][{}] complete scan due total entry <= 0 or last message id is blank, "
+ + "start position is: {}, current result: {}", topic, subscriptionName,
+ startPositionRef.get(), currentResult);
+ future.complete(mergedResult);
+ return;
+ }
+
+ String[] messageIdSplits = mergedResult.getLastMessageId().split(":");
+ MessageIdImpl nextScanMessageId =
+ new MessageIdImpl(Long.parseLong(messageIdSplits[0]), Long.parseLong(messageIdSplits[1]) + 1,
+ partitionIndex);
+ startPositionRef.set(Optional.of(nextScanMessageId));
+
+ resultSupplier.get().whenComplete(this);
+ }
+ };
+
+ resultSupplier.get().whenComplete(completeAction);
+ return future;
+ }
+
+ private AnalyzeSubscriptionBacklogResult mergeBacklogResults(AnalyzeSubscriptionBacklogResult current,
+ AnalyzeSubscriptionBacklogResult previous) {
+ if (previous == null) {
+ return current;
+ }
+
+ AnalyzeSubscriptionBacklogResult mergedRes = new AnalyzeSubscriptionBacklogResult();
+ mergedRes.setEntries(current.getEntries() + previous.getEntries());
+ mergedRes.setMessages(current.getMessages() + previous.getMessages());
+ mergedRes.setMarkerMessages(current.getMarkerMessages() + previous.getMarkerMessages());
+
+ mergedRes.setFilterAcceptedEntries(current.getFilterAcceptedEntries() + previous.getFilterAcceptedEntries());
+ mergedRes.setFilterRejectedEntries(current.getFilterRejectedEntries() + previous.getFilterRejectedEntries());
+ mergedRes.setFilterRescheduledEntries(
+ current.getFilterRescheduledEntries() + previous.getFilterRescheduledEntries());
+
+ mergedRes.setFilterAcceptedMessages(current.getFilterAcceptedMessages() + previous.getFilterAcceptedMessages());
+ mergedRes.setFilterRejectedMessages(current.getFilterRejectedMessages() + previous.getFilterRejectedMessages());
+ mergedRes.setFilterRescheduledMessages(
+ current.getFilterRescheduledMessages() + previous.getFilterRescheduledMessages());
+
+ mergedRes.setAborted(current.isAborted());
+ mergedRes.setFirstMessageId(previous.getFirstMessageId());
+ String lastMessageId = current.getLastMessageId();
+ if (StringUtils.isNotBlank(lastMessageId)) {
+ mergedRes.setLastMessageId(lastMessageId);
+ }
+
+ return mergedRes;
+ }
+
@Override
public Long getBacklogSizeByMessageId(String topic, MessageId messageId)
throws PulsarAdminException {
(pulsar) 01/05: [fix][sec] Upgrade OpenSearch to 2.19.4 to remediate CVE-2025-9624 (#25206)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 05157e71d63c1d8cd5cfc370dedc18a2c6190227
Author: guptas6est <sa...@est.tech>
AuthorDate: Fri Feb 6 15:31:09 2026 +0530
[fix][sec] Upgrade OpenSearch to 2.19.4 to remediate CVE-2025-9624 (#25206)
(cherry picked from commit 64333094705769640d747b868017b01931b17705)
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 70dfa025be8..1cff3a7d0ef 100644
--- a/pom.xml
+++ b/pom.xml
@@ -245,7 +245,7 @@ flexible messaging model and an intuitive client API.</description>
<mariadb-jdbc.version>3.5.5</mariadb-jdbc.version>
<openmldb-jdbc.version>0.4.4-hotfix1</openmldb-jdbc.version>
<json-smart.version>2.5.2</json-smart.version>
- <opensearch.version>2.16.0</opensearch.version>
+ <opensearch.version>2.19.4</opensearch.version>
<elasticsearch-java.version>8.15.3</elasticsearch-java.version>
<debezium.version>3.2.5.Final</debezium.version>
<debezium.postgresql.version>${postgresql-jdbc.version}</debezium.postgresql.version>
(pulsar) 04/05: [improve][broker] Change log level from warn to debug when cursor mark-deleted position ledger doesn't exist (#25200)
Posted by lh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 7d44375b20f725dbc8a3169bd623915b2152f14f
Author: Yike Xiao <km...@live.com>
AuthorDate: Fri Feb 6 19:33:01 2026 +0800
[improve][broker] Change log level from warn to debug when cursor mark-deleted position ledger doesn't exist (#25200)
(cherry picked from commit 00a65e331724973ce963ea38d1c0f7edbcb0d3c4)
---
.../java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
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 27ea422ef6a..fdab798a8a6 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
@@ -2726,7 +2726,8 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
log.debug("No need to reset cursor: {}, current ledger is the last ledger.", cursor);
}
} else {
- log.warn("Cursor: {} does not exist in the managed-ledger.", cursor);
+ // It's possible that the ledger at the mark-deleted position is empty and has been deleted already.
+ log.debug("Cursor: {} mark-deleted position ledger does not exist in the managed-ledger.", cursor);
}
int compareResult = lastAckedPosition.compareTo(markDeletedPosition);