You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2025/09/10 05:05:30 UTC
(pulsar) branch branch-4.1 updated (eb39530e711 -> cc2b177dd66)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a change to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
from eb39530e711 [fix][sec] Upgrade Netty to 4.1.127.Final to address CVEs (#24717)
new 6509ee3e8fd [improve][broker] Reduce unnecessary MessageMetadata parsing by caching the parsed instance in the broker cache (#24682)
new a2eb1aa42e2 [improve][io] Upgrade to Debezium 3.2.2 (#24712)
new cc2b177dd66 [fix][broker] Fix memory leak when metrics are updated in a thread other than FastThreadLocalThread (#24719)
The 3 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:
.../java/org/apache/bookkeeper/mledger/Entry.java | 25 +++
.../apache/bookkeeper/mledger/impl/EntryImpl.java | 24 +++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 11 +-
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 30 +--
.../mledger/impl/ShadowManagedLedgerImpl.java | 5 +
.../mledger/impl/cache/EntryCacheDisabled.java | 3 +-
.../bookkeeper/mledger/impl/cache/RangeCache.java | 17 +-
.../mledger/impl/cache/RangeCacheEntryWrapper.java | 34 +++-
.../mledger/impl/cache/RangeEntryCacheImpl.java | 1 +
pom.xml | 13 +-
.../broker/admin/impl/PersistentTopicsBase.java | 28 ++-
.../intercept/ManagedLedgerInterceptorImpl.java | 12 +-
.../broker/service/AbstractBaseDispatcher.java | 28 ++-
.../pulsar/broker/service/EntryAndMetadata.java | 12 +-
.../apache/pulsar/broker/service/ServerCnx.java | 5 +-
.../NonPersistentDispatcherMultipleConsumers.java | 3 +-
...onPersistentDispatcherSingleActiveConsumer.java | 3 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 4 +-
.../PersistentDispatcherMultipleConsumers.java | 6 +-
.../PersistentDispatcherSingleActiveConsumer.java | 2 +-
.../persistent/PersistentMessageFinder.java | 3 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 6 +-
...tickyKeyDispatcherMultipleConsumersClassic.java | 2 +-
.../service/persistent/PersistentSubscription.java | 12 +-
.../broker/service/persistent/PersistentTopic.java | 24 ++-
.../metrics/DataSketchesOpStatsLogger.java | 87 +--------
.../metrics/DataSketchesSummaryLogger.java | 49 +----
.../prometheus/metrics/ThreadLocalAccessor.java | 135 +++++++++++++
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 2 +-
.../buffer/impl/TopicTransactionBuffer.java | 7 +-
.../service/PersistentMessageFinderTest.java | 2 +-
.../metrics/ThreadLocalAccessorTest.java | 93 +++++++++
.../apache/pulsar/client/impl/MessageImplTest.java | 3 +-
.../apache/pulsar/common/protocol/Commands.java | 208 +++++++++++++++------
.../apache/pulsar/io/debezium/DebeziumSource.java | 20 +-
...tabaseHistory.java => PulsarSchemaHistory.java} | 48 ++---
.../org/apache/pulsar/io/debezium/SerDeUtils.java | 2 +-
...storyTest.java => PulsarSchemaHistoryTest.java} | 40 ++--
.../io/debezium/mongodb/DebeziumMongoDbSource.java | 7 +
.../resources/debezium-mongodb-source-config.yaml | 5 +-
.../io/debezium/mssql/DebeziumMsSqlSource.java | 7 +
.../resources/debezium-mssql-source-config.yaml | 7 +-
pulsar-io/debezium/mysql/pom.xml | 4 +-
.../io/debezium/mysql/DebeziumMysqlSource.java | 7 +
.../resources/debezium-mysql-source-config.yaml | 7 +-
.../io/debezium/oracle/DebeziumOracleSource.java | 7 +
.../resources/debezium-oracle-source-config.yaml | 5 +-
pulsar-io/debezium/pom.xml | 40 ----
.../debezium/postgres/DebeziumPostgresSource.java | 7 +
.../resources/debezium-postgres-source-config.yaml | 7 +-
pulsar-io/mongo/pom.xml | 2 +-
.../apache/pulsar/io/mongodb/MongoSinkTest.java | 19 +-
.../apache/pulsar/io/mongodb/MongoSourceTest.java | 12 +-
src/owasp-dependency-check-suppressions.xml | 43 -----
tests/integration/pom.xml | 2 +-
.../containers/DebeziumMongoDbContainer.java | 2 +-
.../containers/DebeziumMySQLContainer.java | 2 +-
.../containers/DebeziumPostgreSqlContainer.java | 2 +-
.../io/sources/PulsarIOSourceRunner.java | 15 ++
.../tests/integration/io/sources/SourceTester.java | 2 +
.../debezium/DebeziumMongoDbSourceTester.java | 22 ++-
.../debezium/DebeziumMsSqlSourceTester.java | 14 +-
.../debezium/DebeziumMySqlSourceTester.java | 7 +-
.../debezium/DebeziumOracleDbSourceTester.java | 6 +-
.../debezium/DebeziumPostgreSqlSourceTester.java | 10 +-
.../debezium/PulsarDebeziumSourcesTest.java | 4 +-
tests/scripts/pre-integ-tests.sh | 4 +-
67 files changed, 823 insertions(+), 464 deletions(-)
create mode 100644 pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java
create mode 100644 pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java
rename pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/{PulsarDatabaseHistory.java => PulsarSchemaHistory.java} (86%)
rename pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/{PulsarDatabaseHistoryTest.java => PulsarSchemaHistoryTest.java} (87%)
(pulsar) 03/03: [fix][broker] Fix memory leak when metrics are updated in a thread other than FastThreadLocalThread (#24719)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit cc2b177dd66132c0b98d2ca12b7d66453d122cea
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Sep 10 00:12:36 2025 +0800
[fix][broker] Fix memory leak when metrics are updated in a thread other than FastThreadLocalThread (#24719)
---
.../metrics/DataSketchesOpStatsLogger.java | 87 ++-----------
.../metrics/DataSketchesSummaryLogger.java | 49 +-------
.../prometheus/metrics/ThreadLocalAccessor.java | 135 +++++++++++++++++++++
.../metrics/ThreadLocalAccessorTest.java | 93 ++++++++++++++
4 files changed, 239 insertions(+), 125 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java
index 12f54ba48a4..8973ba6a25c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesOpStatsLogger.java
@@ -19,15 +19,10 @@
package org.apache.pulsar.broker.stats.prometheus.metrics;
import com.yahoo.sketches.quantiles.DoublesSketch;
-import com.yahoo.sketches.quantiles.DoublesSketchBuilder;
import com.yahoo.sketches.quantiles.DoublesUnion;
import com.yahoo.sketches.quantiles.DoublesUnionBuilder;
-import io.netty.util.concurrent.FastThreadLocal;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
-import java.util.concurrent.locks.StampedLock;
import org.apache.bookkeeper.stats.OpStatsData;
import org.apache.bookkeeper.stats.OpStatsLogger;
@@ -65,15 +60,7 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
failCountAdder.increment();
failSumAdder.add((long) valueMillis);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.failSketch.update(valueMillis);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateFail(valueMillis);
}
@Override
@@ -82,45 +69,21 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
successCountAdder.increment();
successSumAdder.add((long) valueMillis);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.successSketch.update(valueMillis);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateSuccess(valueMillis);
}
@Override
public void registerSuccessfulValue(long value) {
successCountAdder.increment();
successSumAdder.add(value);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.successSketch.update(value);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateSuccess(value);
}
@Override
public void registerFailedValue(long value) {
failCountAdder.increment();
failSumAdder.add(value);
-
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.failSketch.update(value);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateFail(value);
}
@Override
@@ -141,21 +104,11 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
current = replacement;
replacement = local;
- final DoublesUnion aggregateSuccesss = new DoublesUnionBuilder().build();
+ final DoublesUnion aggregateSuccess = new DoublesUnionBuilder().build();
final DoublesUnion aggregateFail = new DoublesUnionBuilder().build();
- local.map.forEach((localData, b) -> {
- long stamp = localData.lock.writeLock();
- try {
- aggregateSuccesss.update(localData.successSketch);
- localData.successSketch.reset();
- aggregateFail.update(localData.failSketch);
- localData.failSketch.reset();
- } finally {
- localData.lock.unlockWrite(stamp);
- }
- });
-
- successResult = aggregateSuccesss.getResultAndReset();
+ local.record(aggregateSuccess, aggregateFail);
+
+ successResult = aggregateSuccess.getResultAndReset();
failResult = aggregateFail.getResultAndReset();
}
@@ -171,28 +124,4 @@ public class DataSketchesOpStatsLogger implements OpStatsLogger {
DoublesSketch s = success ? successResult : failResult;
return s != null ? s.getQuantile(quantile) : Double.NaN;
}
-
- private static class LocalData {
- private final DoublesSketch successSketch = new DoublesSketchBuilder().build();
- private final DoublesSketch failSketch = new DoublesSketchBuilder().build();
- private final StampedLock lock = new StampedLock();
- }
-
- private static class ThreadLocalAccessor {
- private final Map<LocalData, Boolean> map = new ConcurrentHashMap<>();
- private final FastThreadLocal<LocalData> localData = new FastThreadLocal<LocalData>() {
-
- @Override
- protected LocalData initialValue() throws Exception {
- LocalData localData = new LocalData();
- map.put(localData, Boolean.TRUE);
- return localData;
- }
-
- @Override
- protected void onRemoval(LocalData value) throws Exception {
- map.remove(value);
- }
- };
- }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
index cc144e0eb69..42c189d4bf3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/DataSketchesSummaryLogger.java
@@ -19,15 +19,10 @@
package org.apache.pulsar.broker.stats.prometheus.metrics;
import com.yahoo.sketches.quantiles.DoublesSketch;
-import com.yahoo.sketches.quantiles.DoublesSketchBuilder;
import com.yahoo.sketches.quantiles.DoublesUnion;
import com.yahoo.sketches.quantiles.DoublesUnionBuilder;
-import io.netty.util.concurrent.FastThreadLocal;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
-import java.util.concurrent.locks.StampedLock;
public class DataSketchesSummaryLogger {
@@ -55,14 +50,7 @@ public class DataSketchesSummaryLogger {
countAdder.increment();
sumAdder.add((long) valueMillis);
- LocalData localData = current.localData.get();
-
- long stamp = localData.lock.readLock();
- try {
- localData.successSketch.update(valueMillis);
- } finally {
- localData.lock.unlockRead(stamp);
- }
+ current.getLocalData().updateSuccess(valueMillis);
}
public void rotateLatencyCollection() {
@@ -72,15 +60,7 @@ public class DataSketchesSummaryLogger {
replacement = local;
final DoublesUnion aggregateValues = new DoublesUnionBuilder().build();
- local.map.forEach((localData, b) -> {
- long stamp = localData.lock.writeLock();
- try {
- aggregateValues.update(localData.successSketch);
- localData.successSketch.reset();
- } finally {
- localData.lock.unlockWrite(stamp);
- }
- });
+ local.record(aggregateValues, null);
values = aggregateValues.getResultAndReset();
}
@@ -97,27 +77,4 @@ public class DataSketchesSummaryLogger {
DoublesSketch s = values;
return s != null ? s.getQuantile(quantile) : Double.NaN;
}
-
- private static class LocalData {
- private final DoublesSketch successSketch = new DoublesSketchBuilder().build();
- private final StampedLock lock = new StampedLock();
- }
-
- private static class ThreadLocalAccessor {
- private final Map<LocalData, Boolean> map = new ConcurrentHashMap<>();
- private final FastThreadLocal<LocalData> localData = new FastThreadLocal<LocalData>() {
-
- @Override
- protected LocalData initialValue() throws Exception {
- LocalData localData = new LocalData();
- map.put(localData, Boolean.TRUE);
- return localData;
- }
-
- @Override
- protected void onRemoval(LocalData value) throws Exception {
- map.remove(value);
- }
- };
- }
-}
\ No newline at end of file
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java
new file mode 100644
index 00000000000..6a32ce5b905
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessor.java
@@ -0,0 +1,135 @@
+/*
+ * 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.stats.prometheus.metrics;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.yahoo.sketches.quantiles.DoublesSketch;
+import com.yahoo.sketches.quantiles.DoublesSketchBuilder;
+import com.yahoo.sketches.quantiles.DoublesUnion;
+import io.netty.util.concurrent.FastThreadLocal;
+import io.netty.util.concurrent.FastThreadLocalThread;
+import java.lang.ref.WeakReference;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.StampedLock;
+import org.jspecify.annotations.Nullable;
+
+class ThreadLocalAccessor {
+
+ private final ConcurrentHashMap<LocalData, Boolean> map = new ConcurrentHashMap<>();
+ private final FastThreadLocal<LocalData> localData = new FastThreadLocal<>() {
+
+ @Override
+ protected LocalData initialValue() {
+ LocalData localData = new LocalData(Thread.currentThread());
+ map.put(localData, Boolean.TRUE);
+ return localData;
+ }
+
+ @Override
+ protected void onRemoval(LocalData value) {
+ map.remove(value);
+ }
+ };
+
+ void record(DoublesUnion aggregateSuccess, @Nullable DoublesUnion aggregateFail) {
+ map.keySet().forEach(key -> {
+ key.record(aggregateSuccess, aggregateFail);
+ if (key.shouldRemove()) {
+ map.remove(key);
+ }
+ });
+ }
+
+ LocalData getLocalData() {
+ return localData.get();
+ }
+
+ @VisibleForTesting
+ int getLocalDataCount() {
+ return map.keySet().size();
+ }
+
+ static class LocalData {
+
+ private final DoublesSketch successSketch = new DoublesSketchBuilder().build();
+ private final DoublesSketch failSketch = new DoublesSketchBuilder().build();
+ private final StampedLock lock = new StampedLock();
+ // Keep a weak reference to the owner thread so that we can remove the LocalData when the thread
+ // is not alive anymore or has been garbage collected.
+ // This reference isn't needed when the owner thread is a FastThreadLocalThread and will be null in that case.
+ // The removal is handled by FastThreadLocal#onRemoval when the owner thread is a FastThreadLocalThread.
+ private final WeakReference<Thread> ownerThreadReference;
+
+ LocalData(Thread ownerThread) {
+ if (ownerThread instanceof FastThreadLocalThread) {
+ ownerThreadReference = null;
+ } else {
+ ownerThreadReference = new WeakReference<>(ownerThread);
+ }
+ }
+
+ private boolean shouldRemove() {
+ if (ownerThreadReference == null) {
+ // the owner is a FastThreadLocalThread which handles the removal using FastThreadLocal#onRemoval
+ return false;
+ } else {
+ Thread ownerThread = ownerThreadReference.get();
+ if (ownerThread == null) {
+ // the thread has already been garbage collected, LocalData should be removed
+ return true;
+ } else {
+ // the thread isn't alive anymore, LocalData should be removed
+ return !ownerThread.isAlive();
+ }
+ }
+ }
+
+ void record(DoublesUnion aggregateSuccess, @Nullable DoublesUnion aggregateFail) {
+ long stamp = lock.writeLock();
+ try {
+ aggregateSuccess.update(successSketch);
+ successSketch.reset();
+ if (aggregateFail != null) {
+ aggregateFail.update(failSketch);
+ failSketch.reset();
+ }
+ } finally {
+ lock.unlockWrite(stamp);
+ }
+ }
+
+ void updateSuccess(double value) {
+ long stamp = lock.readLock();
+ try {
+ successSketch.update(value);
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+
+ void updateFail(double value) {
+ long stamp = lock.readLock();
+ try {
+ failSketch.update(value);
+ } finally {
+ lock.unlockRead(stamp);
+ }
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java
new file mode 100644
index 00000000000..94c8337307d
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/prometheus/metrics/ThreadLocalAccessorTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.stats.prometheus.metrics;
+
+import static org.testng.Assert.assertEquals;
+import com.yahoo.sketches.quantiles.DoublesUnion;
+import io.netty.util.concurrent.FastThreadLocalThread;
+import java.util.concurrent.Phaser;
+import org.jspecify.annotations.Nullable;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ThreadLocalAccessorTest {
+
+ @DataProvider
+ public static Object[][] provider() {
+ return new Object[][] {
+ // 1st element: whether the thread is a FastThreadLocalThread
+ // 2nd element: the 2nd argument passed to the `ThreadLocalAccessor#record` method
+ { true, DoublesUnion.builder().build() },
+ { true, null },
+ { false, DoublesUnion.builder().build() },
+ { false, null },
+ };
+ }
+
+ @Test(dataProvider = "provider")
+ public void testShouldRemoveLocalDataWhenOwnerThreadIsNotAlive(
+ boolean fastThreadLocalThread, @Nullable DoublesUnion aggregateFail) throws Exception {
+ // given a ThreadLocalAccessor instance
+ final var threadLocalAccessor = new ThreadLocalAccessor();
+ DoublesUnion aggregateSuccess = DoublesUnion.builder().build();
+ // using phaser to synchronize threads
+ Phaser phaser = new Phaser(2);
+ Thread thread = getThread(fastThreadLocalThread, () -> {
+ // Create a new LocalData instance for the current thread.
+ threadLocalAccessor.getLocalData();
+ // sync point #1, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ // sync point #2, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ });
+ // sync point #1, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ // and record is called
+ threadLocalAccessor.record(aggregateSuccess, aggregateFail);
+ // then LocalData should exist
+ assertEquals(threadLocalAccessor.getLocalDataCount(), 1);
+
+ // when thread is not alive anymore
+ // sync point #2, wait and advance at the same time
+ phaser.arriveAndAwaitAdvance();
+ // wait for thread to finish
+ thread.join();
+ // and record is called
+ threadLocalAccessor.record(aggregateSuccess, aggregateFail);
+ // then LocalData should be removed
+ assertEquals(threadLocalAccessor.getLocalDataCount(), 0);
+ }
+
+ @Test(dataProvider = "provider")
+ public void testThreadGc(boolean fastThreadLocalThread, @Nullable DoublesUnion aggregateFail) throws Exception {
+ final var accessor = new ThreadLocalAccessor();
+ getThread(fastThreadLocalThread, accessor::getLocalData).join();
+ System.gc();
+ // FastThreadLocalThread removes the LocalData from the map when the thread finishes
+ assertEquals(accessor.getLocalDataCount(), fastThreadLocalThread ? 0 : 1);
+ accessor.record(DoublesUnion.builder().build(), aggregateFail);
+ assertEquals(accessor.getLocalDataCount(), 0);
+ }
+
+ private static Thread getThread(boolean fastThreadLocalThread, Runnable runnable) {
+ final var thread = fastThreadLocalThread ? new FastThreadLocalThread(runnable) : new Thread(runnable);
+ thread.start(); // when LocalData is created
+ return thread;
+ }
+}
(pulsar) 01/03: [improve][broker] Reduce unnecessary MessageMetadata parsing by caching the parsed instance in the broker cache (#24682)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 6509ee3e8fd34da26d4dcbf2b438603ab606e879
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Sep 10 00:23:17 2025 +0300
[improve][broker] Reduce unnecessary MessageMetadata parsing by caching the parsed instance in the broker cache (#24682)
---
.../java/org/apache/bookkeeper/mledger/Entry.java | 25 +++
.../apache/bookkeeper/mledger/impl/EntryImpl.java | 24 +++
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 11 +-
.../apache/bookkeeper/mledger/impl/OpAddEntry.java | 30 +--
.../mledger/impl/ShadowManagedLedgerImpl.java | 5 +
.../mledger/impl/cache/EntryCacheDisabled.java | 3 +-
.../bookkeeper/mledger/impl/cache/RangeCache.java | 17 +-
.../mledger/impl/cache/RangeCacheEntryWrapper.java | 34 +++-
.../mledger/impl/cache/RangeEntryCacheImpl.java | 1 +
.../broker/admin/impl/PersistentTopicsBase.java | 28 ++-
.../intercept/ManagedLedgerInterceptorImpl.java | 12 +-
.../broker/service/AbstractBaseDispatcher.java | 28 ++-
.../pulsar/broker/service/EntryAndMetadata.java | 12 +-
.../apache/pulsar/broker/service/ServerCnx.java | 5 +-
.../NonPersistentDispatcherMultipleConsumers.java | 3 +-
...onPersistentDispatcherSingleActiveConsumer.java | 3 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 4 +-
.../PersistentDispatcherMultipleConsumers.java | 6 +-
.../PersistentDispatcherSingleActiveConsumer.java | 2 +-
.../persistent/PersistentMessageFinder.java | 3 +-
...istentStickyKeyDispatcherMultipleConsumers.java | 6 +-
...tickyKeyDispatcherMultipleConsumersClassic.java | 2 +-
.../service/persistent/PersistentSubscription.java | 12 +-
.../broker/service/persistent/PersistentTopic.java | 24 ++-
.../SnapshotSegmentAbortedTxnProcessorImpl.java | 2 +-
.../buffer/impl/TopicTransactionBuffer.java | 7 +-
.../service/PersistentMessageFinderTest.java | 2 +-
.../apache/pulsar/client/impl/MessageImplTest.java | 3 +-
.../apache/pulsar/common/protocol/Commands.java | 208 +++++++++++++++------
29 files changed, 382 insertions(+), 140 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/Entry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/Entry.java
index 6b8f8a95773..24ea5c17c0d 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/Entry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/Entry.java
@@ -21,6 +21,8 @@ package org.apache.bookkeeper.mledger;
import io.netty.buffer.ByteBuf;
import org.apache.bookkeeper.common.annotation.InterfaceAudience;
import org.apache.bookkeeper.common.annotation.InterfaceStability;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
/**
* An Entry represent a ledger entry data and its associated position.
@@ -98,4 +100,27 @@ public interface Entry {
default boolean matchesPosition(Position position) {
return position != null && position.compareTo(getLedgerId(), getEntryId()) == 0;
}
+
+ default MessageMetadata getMessageMetadata() {
+ return null;
+ }
+
+ /**
+ * Returns the timestamp of the entry.
+ * @return
+ */
+ default long getEntryTimestamp() {
+ // get broker timestamp first if BrokerEntryMetadata is enabled with AppendBrokerTimestampMetadataInterceptor
+ return Commands.peekBrokerEntryMetadataToLong(getDataBuffer(), brokerEntryMetadata -> {
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ return brokerEntryMetadata.getBrokerTimestamp();
+ }
+ // otherwise get the publish_time
+ MessageMetadata messageMetadata = getMessageMetadata();
+ if (messageMetadata == null) {
+ messageMetadata = Commands.peekMessageMetadata(getDataBuffer(), null, -1);
+ }
+ return messageMetadata.getPublishTime();
+ });
+ }
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
index b85a1bc45fe..070a0fc1bea 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java
@@ -24,6 +24,9 @@ import io.netty.buffer.Unpooled;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
import io.netty.util.ReferenceCounted;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.api.LedgerEntry;
import org.apache.bookkeeper.client.impl.LedgerEntryImpl;
import org.apache.bookkeeper.mledger.Entry;
@@ -33,7 +36,10 @@ import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.bookkeeper.mledger.ReferenceCountedEntry;
import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.bookkeeper.mledger.util.AbstractCASReferenceCounted;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
+@Slf4j
public final class EntryImpl extends AbstractCASReferenceCounted
implements ReferenceCountedEntry, Comparable<EntryImpl> {
@@ -51,6 +57,8 @@ public final class EntryImpl extends AbstractCASReferenceCounted
ByteBuf data;
private EntryReadCountHandler readCountHandler;
private boolean decreaseReadCountOnRelease = true;
+ @Getter @Setter
+ private MessageMetadata messageMetadata;
private Runnable onDeallocate;
@@ -161,6 +169,7 @@ public final class EntryImpl extends AbstractCASReferenceCounted
entry.entryId = other.entryId;
entry.data = other.data.retainedDuplicate();
entry.readCountHandler = other.readCountHandler;
+ entry.messageMetadata = other.messageMetadata;
entry.setRefCnt(1);
return entry;
}
@@ -172,6 +181,7 @@ public final class EntryImpl extends AbstractCASReferenceCounted
entry.entryId = other.getEntryId();
entry.data = other.getDataBuffer().retainedDuplicate();
entry.readCountHandler = other.getReadCountHandler();
+ entry.messageMetadata = other.getMessageMetadata();
entry.setRefCnt(1);
return entry;
}
@@ -277,6 +287,7 @@ public final class EntryImpl extends AbstractCASReferenceCounted
position = null;
readCountHandler = null;
decreaseReadCountOnRelease = true;
+ messageMetadata = null;
recyclerHandle.recycle(this);
}
@@ -294,6 +305,19 @@ public final class EntryImpl extends AbstractCASReferenceCounted
decreaseReadCountOnRelease = enabled;
}
+ public void initializeMessageMetadataIfNeeded(String managedLedgerName) {
+ if (messageMetadata == null) {
+ try {
+ MessageMetadata msgMetadata = new MessageMetadata();
+ Commands.peekMessageMetadata(data, msgMetadata);
+ this.messageMetadata = msgMetadata;
+ } catch (Throwable t) {
+ log.warn("[{}] Failed to parse message metadata for entry {}:{}", managedLedgerName, ledgerId, entryId,
+ t);
+ }
+ }
+ }
+
@Override
public String toString() {
return getClass().getName() + "@" + System.identityHashCode(this)
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 0be740e1bda..a4527261c8b 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
@@ -31,7 +31,6 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
-import java.io.IOException;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Collection;
@@ -144,7 +143,6 @@ import org.apache.pulsar.common.policies.data.ManagedLedgerInternalStats;
import org.apache.pulsar.common.policies.data.OffloadPolicies;
import org.apache.pulsar.common.policies.data.OffloadedReadPriority;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
-import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.DateFormatter;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.LazyLoadableValue;
@@ -1315,9 +1313,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
@Override
public void readEntryComplete(Entry entry, Object ctx) {
try {
- long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ long entryTimestamp = entry.getEntryTimestamp();
future.complete(entryTimestamp);
- } catch (IOException e) {
+ } catch (Exception e) {
log.error("Error deserializing message for message position {}", nextPos, e);
future.completeExceptionally(e);
} finally {
@@ -4991,4 +4989,9 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
throw new RuntimeException(e);
}
}
+
+ boolean shouldCacheAddedEntry() {
+ // Avoid caching entries if no cursor has been created
+ return getActiveCursors().shouldCacheAddedEntry();
+ }
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
index 49bc3f309d4..3af5618064f 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpAddEntry.java
@@ -257,23 +257,23 @@ public class OpAddEntry implements AddCallback, CloseCallback, Runnable, Managed
ManagedLedgerImpl.NUMBER_OF_ENTRIES_UPDATER.incrementAndGet(ml);
ManagedLedgerImpl.TOTAL_SIZE_UPDATER.addAndGet(ml, dataLength);
+ // ctx will contain a Position instance only in the case of ShadowManagedLedgerImpl
long ledgerId = ledger != null ? ledger.getId() : ((Position) ctx).getLedgerId();
- // Don't insert to the entry cache for the ShadowManagedLedger
- if (!(ml instanceof ShadowManagedLedgerImpl)) {
- // Avoid caching entries if no cursor has been created
- if (ml.getActiveCursors().shouldCacheAddedEntry()) {
- int expectedReadCount = 0;
- // only use expectedReadCount if cache eviction is enabled by expected read count
- if (ml.getConfig().isCacheEvictionByExpectedReadCount()) {
- expectedReadCount = ml.getActiveCursors().size();
- }
- EntryImpl entry = EntryImpl.create(ledgerId, entryId, data, expectedReadCount);
- entry.setDecreaseReadCountOnRelease(false);
- // EntryCache.insert: duplicates entry by allocating new entry and data. so, recycle entry after calling
- // insert
- ml.entryCache.insert(entry);
- entry.release();
+
+ // Handle caching for tailing reads
+ if (ml.shouldCacheAddedEntry()) {
+ int expectedReadCount = 0;
+ // only use expectedReadCount if cache eviction is enabled by expected read count
+ if (ml.getConfig().isCacheEvictionByExpectedReadCount()) {
+ // use the number of active cursors as the expected read count
+ expectedReadCount = ml.getActiveCursors().size();
}
+ EntryImpl entry = EntryImpl.create(ledgerId, entryId, data, expectedReadCount);
+ entry.setDecreaseReadCountOnRelease(false);
+ // EntryCache.insert: duplicates entry by allocating new entry and data. so, recycle entry after calling
+ // insert
+ ml.entryCache.insert(entry);
+ entry.release();
}
Position lastEntry = PositionFactory.create(ledgerId, entryId);
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
index 4b03cad8e0a..46cd1335a17 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ShadowManagedLedgerImpl.java
@@ -407,4 +407,9 @@ public class ShadowManagedLedgerImpl extends ManagedLedgerImpl {
protected void updateLastLedgerCreatedTimeAndScheduleRolloverTask() {
this.lastLedgerCreatedTimestamp = clock.millis();
}
+
+ @Override
+ boolean shouldCacheAddedEntry() {
+ return false;
+ }
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java
index 9cd63d99f4c..b5a45415a4f 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/EntryCacheDisabled.java
@@ -78,6 +78,7 @@ public class EntryCacheDisabled implements EntryCache {
for (LedgerEntry e : ledgerEntries) {
// Insert the entries at the end of the list (they will be unsorted for now)
EntryImpl entry = EntryImpl.create(e, interceptor, 0);
+ entry.initializeMessageMetadataIfNeeded(ml.getName());
entries.add(entry);
totalSize += entry.getLength();
}
@@ -111,7 +112,7 @@ public class EntryCacheDisabled implements EntryCache {
if (iterator.hasNext()) {
LedgerEntry ledgerEntry = iterator.next();
EntryImpl returnEntry = EntryImpl.create(ledgerEntry, interceptor, 0);
-
+ returnEntry.initializeMessageMetadataIfNeeded(ml.getName());
ml.getMbean().recordReadEntriesOpsCacheMisses(1, returnEntry.getLength());
ml.getFactory().getMbean().recordCacheMiss(1, returnEntry.getLength());
ml.getMbean().addReadEntriesSample(1, returnEntry.getLength());
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
index a5592f7098e..858b7d3c07c 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCache.java
@@ -46,13 +46,23 @@ import org.apache.commons.lang3.tuple.Pair;
class RangeCache {
private final ConcurrentNavigableMap<Position, RangeCacheEntryWrapper> entries;
private final RangeCacheRemovalQueue removalQueue;
- private AtomicLong size; // Total size of values stored in cache
+ private final AtomicLong size; // Total size of values stored in cache
+ private final String managedLedgerName;
/**
* Construct a new RangeCache.
*/
public RangeCache(RangeCacheRemovalQueue removalQueue) {
+ this(removalQueue, null);
+ }
+
+ /**
+ * Construct a new RangeCache.
+ * @param managedLedgerName the name of the managed ledger this cache belongs to
+ */
+ public RangeCache(RangeCacheRemovalQueue removalQueue, String managedLedgerName) {
this.removalQueue = removalQueue;
+ this.managedLedgerName = managedLedgerName;
this.entries = new ConcurrentSkipListMap<>();
this.size = new AtomicLong(0);
}
@@ -115,7 +125,7 @@ class RangeCache {
if (valueWrapper == null) {
return null;
} else {
- ReferenceCountedEntry value = valueWrapper.getValue(key);
+ ReferenceCountedEntry value = valueWrapper.getValue(key, managedLedgerName);
return getRetainedValueMatchingKey(key, value);
}
}
@@ -124,7 +134,8 @@ class RangeCache {
* @apiNote the returned value must be released if it's not null
*/
private ReferenceCountedEntry getValueMatchingEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry) {
- ReferenceCountedEntry valueMatchingEntry = RangeCacheEntryWrapper.getValueMatchingMapEntry(entry);
+ ReferenceCountedEntry valueMatchingEntry =
+ RangeCacheEntryWrapper.getValueMatchingMapEntry(entry, managedLedgerName);
return getRetainedValueMatchingKey(entry.getKey(), valueMatchingEntry);
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
index db5ceaefd38..5c82207e1c7 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeCacheEntryWrapper.java
@@ -22,14 +22,17 @@ import io.netty.util.Recycler;
import java.util.Map;
import java.util.concurrent.locks.StampedLock;
import java.util.function.Function;
+import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.ReferenceCountedEntry;
+import org.apache.bookkeeper.mledger.impl.EntryImpl;
/**
* Wrapper around the value to store in Map. This is needed to ensure that a specific instance can be removed from
* the map by calling the {@link Map#remove(Object, Object)} method. Certain race conditions could result in the
* wrong value being removed from the map. The instances of this class are recycled to avoid creating new objects.
*/
+@Slf4j
class RangeCacheEntryWrapper {
private final Recycler.Handle<RangeCacheEntryWrapper> recyclerHandle;
private static final Recycler<RangeCacheEntryWrapper> RECYCLER = new Recycler<RangeCacheEntryWrapper>() {
@@ -73,12 +76,13 @@ class RangeCacheEntryWrapper {
/**
* Get the value associated with the key. Returns null if the key does not match the key.
*
- * @param key the key to match
+ * @param key the key to match
+ * @param managedLedgerName
* @return the value associated with the key, or null if the value has already been recycled or the key does not
* match
*/
- ReferenceCountedEntry getValue(Position key) {
- return getValueInternal(key, false);
+ ReferenceCountedEntry getValue(Position key, String managedLedgerName) {
+ return getValueInternal(key, false, managedLedgerName);
}
/**
@@ -88,8 +92,9 @@ class RangeCacheEntryWrapper {
* @return the value associated with the key, or null if the value has already been recycled or the key does not
* exactly match the same instance
*/
- static ReferenceCountedEntry getValueMatchingMapEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry) {
- return entry.getValue().getValueInternal(entry.getKey(), true);
+ static ReferenceCountedEntry getValueMatchingMapEntry(Map.Entry<Position, RangeCacheEntryWrapper> entry,
+ String managedLedgerName) {
+ return entry.getValue().getValueInternal(entry.getKey(), true, managedLedgerName);
}
/**
@@ -101,16 +106,20 @@ class RangeCacheEntryWrapper {
* key as the one stored in the wrapper. This is used to avoid any races
* when retrieving or removing the entries from the cache when the key and value
* instances are available.
+ * @param managedLedgerName
* @return the value associated with the key, or null if the key does not match
*/
- private ReferenceCountedEntry getValueInternal(Position key, boolean requireSameKeyInstance) {
+ private ReferenceCountedEntry getValueInternal(Position key, boolean requireSameKeyInstance,
+ String managedLedgerName) {
long stamp = lock.tryOptimisticRead();
Position localKey = this.key;
ReferenceCountedEntry localValue = this.value;
+ boolean messageMetadataInitialized = localValue != null && localValue.getMessageMetadata() != null;
if (!lock.validate(stamp)) {
stamp = lock.readLock();
localKey = this.key;
localValue = this.value;
+ messageMetadataInitialized = localValue != null && localValue.getMessageMetadata() != null;
lock.unlockRead(stamp);
}
// check that the given key matches the key associated with the value in the entry
@@ -120,6 +129,19 @@ class RangeCacheEntryWrapper {
if (localKey != key && (requireSameKeyInstance || localKey == null || !localKey.equals(key))) {
return null;
}
+ // Initialize the metadata if it's not already initialized
+ if (localValue != null && !messageMetadataInitialized) {
+ localValue = withWriteLock(wrapper -> {
+ // ensure that the key still matches
+ if (wrapper.key != key && (requireSameKeyInstance || wrapper.key == null || !wrapper.key.equals(key))) {
+ return null;
+ }
+ if (wrapper.value instanceof EntryImpl entry) {
+ entry.initializeMessageMetadataIfNeeded(managedLedgerName);
+ }
+ return wrapper.value;
+ });
+ }
accessed = true;
return localValue;
}
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
index 04de8dde0ad..fd391ba2bf6 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/cache/RangeEntryCacheImpl.java
@@ -532,6 +532,7 @@ public class RangeEntryCacheImpl implements EntryCache {
final List<Entry> entriesToReturn = new ArrayList<>(entriesToRead);
for (LedgerEntry e : ledgerEntries) {
EntryImpl entry = EntryImpl.create(e, interceptor, expectedReadCountVal);
+ entry.initializeMessageMetadataIfNeeded(ml.getName());
entriesToReturn.add(entry);
totalSize += entry.getLength();
if (expectedReadCountVal > 0) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index a16be37bb38..ec61d58d2af 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -2659,8 +2659,10 @@ public class PersistentTopicsBase extends AdminResource {
if (entry == null) {
batchSizeFuture.complete(0);
} else {
- MessageMetadata metadata =
- Commands.parseMessageMetadata(entry.getDataBuffer());
+ MessageMetadata metadata = entry.getMessageMetadata();
+ if (metadata == null) {
+ metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ }
batchSizeFuture.complete(metadata.getNumMessagesInBatch());
}
} catch (Exception e) {
@@ -2839,7 +2841,7 @@ public class PersistentTopicsBase extends AdminResource {
private CompletableFuture<MessageId> findMessageIdByPublishTime(long timestamp, ManagedLedger managedLedger) {
return managedLedger.asyncFindPosition(entry -> {
try {
- long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ long entryTimestamp = entry.getEntryTimestamp();
return MessageImpl.isEntryPublishedEarlierThan(entryTimestamp, timestamp);
} catch (Exception e) {
log.error("[{}] Error deserializing message for message position find",
@@ -3009,7 +3011,12 @@ public class PersistentTopicsBase extends AdminResource {
long totalSize = metadataAndPayload.readableBytes();
BrokerEntryMetadata brokerEntryMetadata = Commands.peekBrokerEntryMetadataIfExist(metadataAndPayload);
- MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload);
+ MessageMetadata metadata = entry.getMessageMetadata();
+ if (metadata == null) {
+ metadata = Commands.parseMessageMetadata(metadataAndPayload);
+ } else {
+ Commands.skipMessageMetadata(metadataAndPayload);
+ }
ResponseBuilder responseBuilder = Response.ok();
responseBuilder.header("X-Pulsar-Message-ID", pos.toString());
@@ -5451,11 +5458,12 @@ public class PersistentTopicsBase extends AdminResource {
private static Long getIndexFromEntry(Entry entry) {
- final var brokerEntryMetadata = Commands.parseBrokerEntryMetadataIfExist(entry.getDataBuffer());
- if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
- return brokerEntryMetadata.getIndex();
- } else {
- return null;
- }
+ return Commands.peekBrokerEntryMetadataToObject(entry.getDataBuffer(), brokerEntryMetadata -> {
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ return brokerEntryMetadata.getIndex();
+ } else {
+ return null;
+ }
+ });
}
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
index db138989a8e..a86b49d627f 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/ManagedLedgerInterceptorImpl.java
@@ -26,7 +26,6 @@ import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.intercept.ManagedLedgerInterceptor;
import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.pulsar.common.api.proto.BrokerEntryMetadata;
import org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor;
import org.apache.pulsar.common.intercept.BrokerEntryMetadataInterceptor;
import org.apache.pulsar.common.intercept.ManagedLedgerPayloadProcessor;
@@ -117,12 +116,11 @@ public class ManagedLedgerInterceptorImpl implements ManagedLedgerInterceptor {
if (lastEntryOptional.isPresent()) {
Entry lastEntry = lastEntryOptional.get();
try {
- BrokerEntryMetadata brokerEntryMetadata =
- Commands.parseBrokerEntryMetadataIfExist(lastEntry.getDataBuffer());
- if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
- appendIndexMetadataInterceptor.recoveryIndexGenerator(
- brokerEntryMetadata.getIndex());
- }
+ Commands.peekBrokerEntryMetadataAndConsume(lastEntry.getDataBuffer(), brokerEntryMetadata -> {
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasIndex()) {
+ appendIndexMetadataInterceptor.recoveryIndexGenerator(brokerEntryMetadata.getIndex());
+ }
+ });
} finally {
lastEntry.release();
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
index f074f234b87..c5e001692f2 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
@@ -149,6 +149,8 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
msgMetadata = metadataArray[metadataIndex];
} else if (entry instanceof EntryAndMetadata) {
msgMetadata = ((EntryAndMetadata) entry).getMetadata();
+ } else if (entry.getMessageMetadata() != null) {
+ msgMetadata = entry.getMessageMetadata();
} else {
msgMetadata = Commands.peekAndCopyMessageMetadata(metadataAndPayload, subscription.toString(), -1);
}
@@ -454,8 +456,18 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
return true;
}
- protected byte[] peekStickyKey(ByteBuf metadataAndPayload) {
- return Commands.peekStickyKey(metadataAndPayload, subscription.getTopicName(), subscription.getName());
+ protected byte[] peekStickyKey(Entry entry) {
+ if (entry instanceof EntryAndMetadata entryAndMetadata) {
+ return entryAndMetadata.getStickyKey();
+ }
+ MessageMetadata metadata = entry.getMessageMetadata();
+ if (metadata == null) {
+ metadata = Commands.peekMessageMetadata(entry.getDataBuffer(), subscription.toString(), -1);
+ }
+ if (metadata == null) {
+ return Commands.NONE_KEY;
+ }
+ return Commands.resolveStickyKey(metadata);
}
protected String getSubscriptionName() {
@@ -528,4 +540,16 @@ public abstract class AbstractBaseDispatcher extends EntryFilterSupport implemen
protected final void updatePendingBytesToDispatch(long size) {
PENDING_BYTES_TO_DISPATCH.inc(size);
}
+
+ protected int getNumberOfMessagesInBatch(Entry entry) {
+ MessageMetadata msgMetadata = entry.getMessageMetadata();
+ if (msgMetadata == null) {
+ msgMetadata = Commands.peekMessageMetadata(entry.getDataBuffer(), subscription.toString(), -1);
+ }
+ if (msgMetadata == null) {
+ return -1;
+ } else {
+ return msgMetadata.getNumMessagesInBatch();
+ }
+ }
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java
index 33abddc300b..117f0115064 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java
@@ -43,12 +43,22 @@ public class EntryAndMetadata implements Entry {
}
public static EntryAndMetadata create(final Entry entry, final MessageMetadata metadata) {
+ if (entry instanceof EntryAndMetadata entryAndMetadata) {
+ return entryAndMetadata;
+ }
return new EntryAndMetadata(entry, metadata);
}
@VisibleForTesting
public static EntryAndMetadata create(final Entry entry) {
- return create(entry, Commands.peekAndCopyMessageMetadata(entry.getDataBuffer(), "", -1));
+ if (entry instanceof EntryAndMetadata entryAndMetadata) {
+ return entryAndMetadata;
+ }
+ MessageMetadata msgMetadata = entry.getMessageMetadata();
+ if (msgMetadata == null) {
+ msgMetadata = Commands.peekAndCopyMessageMetadata(entry.getDataBuffer(), "", -1);
+ }
+ return create(entry, msgMetadata);
}
public byte[] getStickyKey() {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
index d1bc4953d11..44927c375b5 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -2365,7 +2365,10 @@ public class ServerCnx extends PulsarHandler implements TransportCnx {
}, null);
CompletableFuture<Integer> batchSizeFuture = entryFuture.thenApply(entry -> {
- MessageMetadata metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ MessageMetadata metadata = entry.getMessageMetadata();
+ if (metadata == null) {
+ metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ }
int batchSize = metadata.getNumMessagesInBatch();
entry.release();
return metadata.hasNumMessagesInBatch() ? batchSize : -1;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
index 5941093e71c..a4fba30cc9b 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java
@@ -37,7 +37,6 @@ import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
-import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.stats.Rate;
import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
@@ -205,7 +204,7 @@ public class NonPersistentDispatcherMultipleConsumers extends AbstractDispatcher
TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -sendMessageInfo.getTotalMessages());
} else {
entries.forEach(entry -> {
- int totalMsgs = Commands.getNumberOfMessagesInBatch(entry.getDataBuffer(), subscription.toString(), -1);
+ int totalMsgs = getNumberOfMessagesInBatch(entry);
if (totalMsgs > 0) {
msgDrop.recordEvent(totalMsgs);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
index 414e9235418..26bb5a791cf 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherSingleActiveConsumer.java
@@ -30,7 +30,6 @@ import org.apache.pulsar.broker.service.RedeliveryTrackerDisabled;
import org.apache.pulsar.broker.service.SendMessageInfo;
import org.apache.pulsar.broker.service.Subscription;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
-import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.stats.Rate;
@Slf4j
@@ -66,7 +65,7 @@ public final class NonPersistentDispatcherSingleActiveConsumer extends AbstractD
sendMessageInfo.getTotalBytes(), sendMessageInfo.getTotalChunkedMessages(), getRedeliveryTracker());
} else {
entries.forEach(entry -> {
- int totalMsgs = Commands.getNumberOfMessagesInBatch(entry.getDataBuffer(), subscription.toString(), -1);
+ int totalMsgs = getNumberOfMessagesInBatch(entry);
if (totalMsgs > 0) {
msgDrop.recordEvent(totalMsgs);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
index 915f9c0f925..2e884eb0ea1 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java
@@ -151,7 +151,7 @@ public class NonPersistentStickyKeyDispatcherMultipleConsumers extends NonPersis
consumerStickyKeyHashesMap.clear();
for (Entry entry : entries) {
- byte[] stickyKey = peekStickyKey(entry.getDataBuffer());
+ byte[] stickyKey = peekStickyKey(entry);
int stickyKeyHash = selector.makeStickyKeyHash(stickyKey);
Consumer consumer = selector.select(stickyKeyHash);
@@ -182,7 +182,7 @@ public class NonPersistentStickyKeyDispatcherMultipleConsumers extends NonPersis
TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -sendMessageInfo.getTotalMessages());
} else {
entriesForConsumer.forEach(e -> {
- int totalMsgs = Commands.getNumberOfMessagesInBatch(e.getDataBuffer(), subscription.toString(), -1);
+ int totalMsgs = getNumberOfMessagesInBatch(e);
if (totalMsgs > 0) {
msgDrop.recordEvent(totalMsgs);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
index c35d802f43d..097ab9cd0fe 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
@@ -72,7 +72,6 @@ import org.apache.pulsar.broker.transaction.exception.buffer.TransactionBufferEx
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.policies.data.stats.TopicMetricBean;
-import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.FutureUtil;
@@ -798,10 +797,11 @@ public class PersistentDispatcherMultipleConsumers extends AbstractPersistentDis
if (entry instanceof EntryAndMetadata) {
metadata = ((EntryAndMetadata) entry).getMetadata();
} else {
- metadata = Commands.peekAndCopyMessageMetadata(entry.getDataBuffer(), subscription.toString(), -1);
// cache the metadata in the entry with EntryAndMetadata for later use to avoid re-parsing the metadata
// and to carry the metadata and calculated stickyKeyHash with the entry
- entries.set(i, EntryAndMetadata.create(entry, metadata));
+ EntryAndMetadata entryAndMetadata = EntryAndMetadata.create(entry);
+ metadata = entryAndMetadata.getMetadata();
+ entries.set(i, entryAndMetadata);
}
if (metadata != null) {
remainingMessages += metadata.getNumMessagesInBatch();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
index 99a09d3a5d7..e809d984ae3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
@@ -189,7 +189,7 @@ public class PersistentDispatcherSingleActiveConsumer extends AbstractDispatcher
Iterator<Entry> iterator = entries.iterator();
while (iterator.hasNext()) {
Entry entry = iterator.next();
- byte[] key = peekStickyKey(entry.getDataBuffer());
+ byte[] key = peekStickyKey(entry);
Consumer consumer = stickyKeyConsumerSelector.select(key);
// Skip the entry if it's not for current active consumer.
if (consumer == null || currentConsumer != consumer) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java
index e8bc9fdc3ea..991e6a06029 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageFinder.java
@@ -30,7 +30,6 @@ import org.apache.bookkeeper.mledger.PositionFactory;
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedLedgerInfo.LedgerInfo;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.client.impl.MessageImpl;
-import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.Codec;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,7 +72,7 @@ public class PersistentMessageFinder implements AsyncCallbacks.FindEntryCallback
cursor.asyncFindNewestMatching(ManagedCursor.FindPositionConstraint.SearchAllAvailableEntries, entry -> {
try {
// Find the latest entry that is earlier than the target timestamp.
- long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ long entryTimestamp = entry.getEntryTimestamp();
return MessageImpl.isEntryPublishedEarlierThan(entryTimestamp, timestamp);
} catch (Exception e) {
log.error("[{}][{}] Error deserializing message for message position find", topicName, subName, e);
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
index 080f5acbf16..e0c74f9043c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java
@@ -58,7 +58,6 @@ import org.apache.pulsar.client.api.Range;
import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType;
import org.apache.pulsar.common.api.proto.KeySharedMeta;
import org.apache.pulsar.common.api.proto.KeySharedMode;
-import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -419,8 +418,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
} else {
// replace the input entry with EntryAndMetadata instance. In addition to the entry and metadata,
// it will also carry the calculated sticky key hash
- entry = EntryAndMetadata.create(inputEntry,
- Commands.peekAndCopyMessageMetadata(inputEntry.getDataBuffer(), getSubscriptionName(), -1));
+ entry = EntryAndMetadata.create(inputEntry);
}
int stickyKeyHash = getStickyKeyHash(entry);
Consumer consumer = null;
@@ -610,7 +608,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumers extends PersistentDi
// use the cached sticky key hash if available, otherwise calculate the sticky key hash and cache it
return entryAndMetadata.getOrUpdateCachedStickyKeyHash(selector::makeStickyKeyHash);
}
- return selector.makeStickyKeyHash(peekStickyKey(entry.getDataBuffer()));
+ return selector.makeStickyKeyHash(peekStickyKey(entry));
}
@Override
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
index 56161d8dd15..9895a4b35b3 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumersClassic.java
@@ -631,7 +631,7 @@ public class PersistentStickyKeyDispatcherMultipleConsumersClassic
// use the cached sticky key hash if available, otherwise calculate the sticky key hash and cache it
return entryAndMetadata.getOrUpdateCachedStickyKeyHash(selector::makeStickyKeyHash);
}
- return selector.makeStickyKeyHash(peekStickyKey(entry.getDataBuffer()));
+ return selector.makeStickyKeyHash(peekStickyKey(entry));
}
private static final Logger log =
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
index ac54957f0c6..056e9a25fda 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -23,7 +23,6 @@ import static org.apache.pulsar.common.naming.SystemTopicNames.isEventSystemTopi
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import io.netty.buffer.ByteBuf;
-import java.io.IOException;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
@@ -644,7 +643,12 @@ public class PersistentSubscription extends AbstractSubscription {
firstPosition.compareAndSet(null, entryPosition);
lastPosition.set(entryPosition);
ByteBuf metadataAndPayload = entry.getDataBuffer();
- MessageMetadata messageMetadata = Commands.peekMessageMetadata(metadataAndPayload, "", -1);
+ MessageMetadata messageMetadata;
+ if (entry.getMessageMetadata() != null) {
+ messageMetadata = entry.getMessageMetadata();
+ } else {
+ messageMetadata = Commands.peekMessageMetadata(metadataAndPayload, "", -1);
+ }
int numMessages = 1;
if (messageMetadata.hasNumMessagesInBatch()) {
numMessages = messageMetadata.getNumMessagesInBatch();
@@ -1438,9 +1442,9 @@ public class PersistentSubscription extends AbstractSubscription {
@Override
public void readEntryComplete(Entry entry, Object ctx) {
try {
- long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ long entryTimestamp = entry.getEntryTimestamp();
future.complete(entryTimestamp);
- } catch (IOException e) {
+ } catch (Exception e) {
log.error("Error deserializing message for message position {}", nextPos, e);
future.completeExceptionally(e);
} finally {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 657f03d6b38..3bbd8b00777 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -31,7 +31,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import io.netty.buffer.ByteBuf;
import io.netty.util.concurrent.FastThreadLocal;
-import java.io.IOException;
import java.time.Clock;
import java.util.ArrayList;
import java.util.Collections;
@@ -3801,7 +3800,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
@Override
public void readEntryComplete(Entry entry, Object ctx) {
try {
- long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ long entryTimestamp = entry.getEntryTimestamp();
updateResultIfNewer(
new OldestPositionInfo(
oldestMarkDeleteCursorInfo.getPosition(),
@@ -3979,7 +3978,7 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
try {
entry = cursor.getNthEntry(1, IndividualDeletedEntries.Include);
if (entry != null) {
- long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ long entryTimestamp = entry.getEntryTimestamp();
isOldestMessageExpired = MessageImpl.isEntryExpired(
(int) (messageTTLInSeconds * MESSAGE_EXPIRY_THRESHOLD), entryTimestamp);
}
@@ -4009,10 +4008,10 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
public void readEntryComplete(Entry entry, Object ctx) {
long entryTimestamp = 0;
try {
- entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ entryTimestamp = entry.getEntryTimestamp();
res.complete(MessageImpl.isEntryExpired(
(int) (messageTTLInSeconds * MESSAGE_EXPIRY_THRESHOLD), entryTimestamp));
- } catch (IOException e) {
+ } catch (Exception e) {
log.warn("[{}] [{}] Error while getting the oldest message", topic, cursor.toString(), e);
res.complete(false);
}
@@ -4110,7 +4109,10 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
return CompletableFuture.completedFuture(lastDispatchablePosition);
}
return ledger.getLastDispatchablePosition(entry -> {
- MessageMetadata md = Commands.parseMessageMetadata(entry.getDataBuffer());
+ MessageMetadata md = entry.getMessageMetadata();
+ if (md == null) {
+ md = Commands.parseMessageMetadata(entry.getDataBuffer());
+ }
// If a messages has marker will filter by AbstractBaseDispatcher.filterEntriesForConsumer
if (Markers.isServerOnlyMarker(md)) {
return false;
@@ -4179,7 +4181,10 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
@Override
public void readEntryComplete(Entry entry, Object ctx) {
try {
- MessageMetadata metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ MessageMetadata metadata = entry.getMessageMetadata();
+ if (metadata == null) {
+ metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
+ }
if (metadata.hasNumMessagesInBatch()) {
completableFuture.complete(new BatchMessageIdImpl(position.getLedgerId(), position.getEntryId(),
partitionIndex, metadata.getNumMessagesInBatch() - 1));
@@ -4797,7 +4802,10 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
public void readEntryComplete(Entry entry, Object ctx) {
try {
ByteBuf metadataAndPayload = entry.getDataBuffer();
- MessageMetadata msgMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ MessageMetadata msgMetadata = entry.getMessageMetadata();
+ if (msgMetadata == null) {
+ msgMetadata = Commands.parseMessageMetadata(metadataAndPayload);
+ }
long publishTime = msgMetadata.getPublishTime();
future.complete(publishTime);
} catch (Exception e) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
index 98d8a40eb36..38b902dc520 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
@@ -408,7 +408,7 @@ public class SnapshotSegmentAbortedTxnProcessorImpl implements AbortedTxnProcess
//decode snapshot from entry
ByteBuf headersAndPayload = entry.getDataBuffer();
//skip metadata
- Commands.parseMessageMetadata(headersAndPayload);
+ Commands.skipMessageMetadata(headersAndPayload);
TransactionBufferSnapshotSegment snapshotSegment = Schema.AVRO(TransactionBufferSnapshotSegment.class)
.decode(Unpooled.wrappedBuffer(headersAndPayload).nioBuffer());
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
index c43f0ed7fb9..c3961dca6d4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
@@ -184,8 +184,11 @@ public class TopicTransactionBuffer extends TopicTransactionBufferState implemen
public void handleTxnEntry(Entry entry) {
ByteBuf metadataAndPayload = entry.getDataBuffer();
- MessageMetadata msgMetadata = Commands.peekMessageMetadata(metadataAndPayload,
- TopicTransactionBufferRecover.SUBSCRIPTION_NAME, -1);
+ MessageMetadata msgMetadata = entry.getMessageMetadata();
+ if (msgMetadata == null) {
+ msgMetadata = Commands.peekMessageMetadata(metadataAndPayload,
+ TopicTransactionBufferRecover.SUBSCRIPTION_NAME, -1);
+ }
if (msgMetadata != null && msgMetadata.hasTxnidMostBits() && msgMetadata.hasTxnidLeastBits()) {
TxnID txnID = new TxnID(msgMetadata.getTxnidMostBits(), msgMetadata.getTxnidLeastBits());
Position position = PositionFactory.create(entry.getLedgerId(), entry.getEntryId());
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
index 4dc10671e6a..5707a6cca60 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentMessageFinderTest.java
@@ -635,7 +635,7 @@ public class PersistentMessageFinderTest extends MockedBookKeeperTestCase {
private CompletableFuture<MessageId> findMessageIdByPublishTime(long timestamp, ManagedLedger managedLedger) {
return managedLedger.asyncFindPosition(entry -> {
try {
- long entryTimestamp = Commands.getEntryTimestamp(entry.getDataBuffer());
+ long entryTimestamp = entry.getEntryTimestamp();
return MessageImpl.isEntryPublishedEarlierThan(entryTimestamp, timestamp);
} catch (Exception e) {
log.error("Error deserializing message for message position find", e);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java
index b152eda4689..f3eb7b434a9 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java
@@ -28,7 +28,6 @@ import static org.testng.Assert.assertTrue;
import static org.testng.AssertJUnit.fail;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.CompositeByteBuf;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
@@ -480,7 +479,7 @@ public class MessageImplTest {
entryTimestamp = Commands.getEntryTimestamp(compositeByteBuf);
assertFalse(MessageImpl.isEntryExpired(24 * 3600, entryTimestamp));
assertEquals(entryTimestamp, brokerEntryTimestamp);
- } catch (IOException e) {
+ } catch (Exception e) {
fail();
}
}
diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
index 7aaa010c11e..cab4dc8bcab 100644
--- a/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
+++ b/pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java
@@ -22,6 +22,7 @@ import static com.scurrilous.circe.checksum.Crc32cIntChecksum.computeChecksum;
import static com.scurrilous.circe.checksum.Crc32cIntChecksum.resumeChecksum;
import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
import com.google.common.base.Strings;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.CompositeByteBuf;
@@ -36,6 +37,8 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.function.Consumer;
+import java.util.function.ToLongFunction;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
@@ -457,7 +460,7 @@ public class Commands {
public static void skipChecksumIfPresent(ByteBuf buffer) {
if (hasChecksum(buffer)) {
- readChecksum(buffer);
+ buffer.skipBytes(Short.BYTES + Integer.BYTES);
}
}
@@ -488,15 +491,21 @@ public class Commands {
buffer.skipBytes(metadataSize);
}
- public static long getEntryTimestamp(ByteBuf headersAndPayloadWithBrokerEntryMetadata) throws IOException {
+ /**
+ * Gets the entry timestamp from either broker metadata broker timestamp or the message metadata publish time.
+ * Prefer using Managed Ledger's Entry's getEntryTimestamp() method over this method.
+ * @param headersAndPayloadWithBrokerEntryMetadata headers and payload for the message
+ * @return the entry timestamp
+ */
+ public static long getEntryTimestamp(ByteBuf headersAndPayloadWithBrokerEntryMetadata) {
// get broker timestamp first if BrokerEntryMetadata is enabled with AppendBrokerTimestampMetadataInterceptor
- BrokerEntryMetadata brokerEntryMetadata =
- Commands.parseBrokerEntryMetadataIfExist(headersAndPayloadWithBrokerEntryMetadata);
- if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
- return brokerEntryMetadata.getBrokerTimestamp();
- }
- // otherwise get the publish_time
- return parseMessageMetadata(headersAndPayloadWithBrokerEntryMetadata).getPublishTime();
+ return peekBrokerEntryMetadataToLong(headersAndPayloadWithBrokerEntryMetadata, brokerEntryMetadata -> {
+ if (brokerEntryMetadata != null && brokerEntryMetadata.hasBrokerTimestamp()) {
+ return brokerEntryMetadata.getBrokerTimestamp();
+ }
+ // otherwise get the publish_time
+ return parseMessageMetadata(headersAndPayloadWithBrokerEntryMetadata).getPublishTime();
+ });
}
public static BaseCommand newMessageCommand(long consumerId, long ledgerId, long entryId, int partition,
@@ -1761,39 +1770,126 @@ public class Commands {
return compositeByteBuf;
}
- public static ByteBuf skipBrokerEntryMetadataIfExist(ByteBuf headerAndPayloadWithBrokerEntryMetadata) {
- int readerIndex = headerAndPayloadWithBrokerEntryMetadata.readerIndex();
- if (headerAndPayloadWithBrokerEntryMetadata.readShort() == magicBrokerEntryMetadata) {
- int brokerEntryMetadataSize = headerAndPayloadWithBrokerEntryMetadata.readInt();
- headerAndPayloadWithBrokerEntryMetadata.readerIndex(headerAndPayloadWithBrokerEntryMetadata.readerIndex()
- + brokerEntryMetadataSize);
- } else {
- headerAndPayloadWithBrokerEntryMetadata.readerIndex(readerIndex);
+ /**
+ * Moves the readerIndex ahead skipping possible BrokerEntryMetadata if it exists in the header and payload
+ * buffer.
+ * @param headerAndPayload the header and payload buffer
+ * @return the header and payload buffer passed as parameter
+ */
+ public static ByteBuf skipBrokerEntryMetadataIfExist(ByteBuf headerAndPayload) {
+ int readerIndex = headerAndPayload.readerIndex();
+ if (headerAndPayload.getShort(readerIndex) == magicBrokerEntryMetadata) {
+ headerAndPayload.skipBytes(Short.BYTES);
+ int brokerEntryMetadataSize = headerAndPayload.readInt();
+ headerAndPayload.skipBytes(brokerEntryMetadataSize);
}
- return headerAndPayloadWithBrokerEntryMetadata;
+ return headerAndPayload;
+ }
+
+ /**
+ * Parses the broker entry metadata from the header and payload buffer and returns a new BrokerEntryMetadata
+ * instance if the broker entry metadata exists in the header and payload buffer. Null is returned if the
+ * broker entry metadata does not exist in the header and payload buffer.
+ * The readerIndex of the headerAndPayload buffer is advanced.
+ *
+ * @param headerAndPayload the header and payload buffer
+ * @return broker entry metadata or null
+ */
+ public static BrokerEntryMetadata parseBrokerEntryMetadataIfExist(ByteBuf headerAndPayload) {
+ return parseOrPeekBrokerEntryMetadataIfExist(headerAndPayload, null, false);
+ }
+
+ /**
+ * Parses the broker entry metadata from the header and payload buffer and returns a new BrokerEntryMetadata
+ * instance if the broker entry metadata exists in the header and payload buffer. Null is returned if the
+ * broker entry metadata does not exist in the header and payload buffer.
+ * The readerIndex of the headerAndPayload buffer is not advanced.
+ *
+ * @param headerAndPayload the header and payload buffer
+ * @return broker entry metadata or null
+ */
+ public static BrokerEntryMetadata peekBrokerEntryMetadataIfExist(
+ ByteBuf headerAndPayload) {
+ return parseOrPeekBrokerEntryMetadataIfExist(headerAndPayload, null, true);
}
- public static BrokerEntryMetadata parseBrokerEntryMetadataIfExist(
- ByteBuf headerAndPayloadWithBrokerEntryMetadata) {
- int readerIndex = headerAndPayloadWithBrokerEntryMetadata.readerIndex();
- if (headerAndPayloadWithBrokerEntryMetadata.getShort(readerIndex) == magicBrokerEntryMetadata) {
- headerAndPayloadWithBrokerEntryMetadata.skipBytes(2);
- int brokerEntryMetadataSize = headerAndPayloadWithBrokerEntryMetadata.readInt();
- BrokerEntryMetadata brokerEntryMetadata = new BrokerEntryMetadata();
- brokerEntryMetadata.parseFrom(headerAndPayloadWithBrokerEntryMetadata, brokerEntryMetadataSize);
- return brokerEntryMetadata;
+ /**
+ * Internal method for parsing and peeking broker entry metadata.
+ * @param headerAndPayload header and payload buffer
+ * @param brokerEntryMetadata the broker entry metadata instance to reuse, null if a new instance should be created
+ * @param peek when true, the readerIndex of the headerAndPayload buffer is resetted to the original
+ * @return the broker entry metadata instance or null
+ */
+ private static BrokerEntryMetadata parseOrPeekBrokerEntryMetadataIfExist(
+ ByteBuf headerAndPayload, BrokerEntryMetadata brokerEntryMetadata, boolean peek) {
+ int readerIndex = headerAndPayload.readerIndex();
+ if (headerAndPayload.getShort(readerIndex) == magicBrokerEntryMetadata) {
+ headerAndPayload.skipBytes(Short.BYTES);
+ try {
+ int brokerEntryMetadataSize = headerAndPayload.readInt();
+ if (brokerEntryMetadata == null) {
+ brokerEntryMetadata = new BrokerEntryMetadata();
+ }
+ brokerEntryMetadata.parseFrom(headerAndPayload, brokerEntryMetadataSize);
+ return brokerEntryMetadata;
+ } finally {
+ if (peek) {
+ headerAndPayload.readerIndex(readerIndex);
+ }
+ }
} else {
return null;
}
}
- public static BrokerEntryMetadata peekBrokerEntryMetadataIfExist(
- ByteBuf headerAndPayloadWithBrokerEntryMetadata) {
- final int readerIndex = headerAndPayloadWithBrokerEntryMetadata.readerIndex();
- BrokerEntryMetadata entryMetadata =
- parseBrokerEntryMetadataIfExist(headerAndPayloadWithBrokerEntryMetadata);
- headerAndPayloadWithBrokerEntryMetadata.readerIndex(readerIndex);
- return entryMetadata;
+ /**
+ * Peeks the BrokerEntryMetadata from the given payload and applies the function to the result.
+ * null will be passed to the function if no BrokerEntryMetadata is found.
+ * The function shouldn't return the BrokerEntryMetadata instance or reference it after the function completes
+ * since it's a ThreadLocal instance that is reused.
+ *
+ * @param headerAndPayload the header and payload of the message
+ * @param function the function to apply to the BrokerEntryMetadata
+ * @param <T> the return type of the function
+ * @return the result of the function
+ */
+ public static <T> T peekBrokerEntryMetadataToObject(ByteBuf headerAndPayload,
+ Function<BrokerEntryMetadata, T> function) {
+ BrokerEntryMetadata brokerEntryMetadata =
+ parseOrPeekBrokerEntryMetadataIfExist(headerAndPayload, BROKER_ENTRY_METADATA.get(), true);
+ return function.apply(brokerEntryMetadata);
+ }
+
+ /**
+ * Peeks the BrokerEntryMetadata from the given payload and applies a function returning a long value to the result.
+ * null will be passed to the function if no BrokerEntryMetadata is found. The function shouldn't reference the
+ * BrokerEntryMetadata instance after the function completes since it's a ThreadLocal instance that is reused.
+ *
+ * @param headerAndPayload the header and payload of the message
+ * @param function the function to apply to the BrokerEntryMetadata
+ * @return the result of the function
+ */
+ public static long peekBrokerEntryMetadataToLong(ByteBuf headerAndPayload,
+ ToLongFunction<BrokerEntryMetadata> function) {
+ BrokerEntryMetadata brokerEntryMetadata =
+ parseOrPeekBrokerEntryMetadataIfExist(headerAndPayload, BROKER_ENTRY_METADATA.get(), true);
+ return function.applyAsLong(brokerEntryMetadata);
+ }
+
+ /**
+ * Peeks the BrokerEntryMetadata from the given payload and consumes the value using a function.
+ * null will be passed to the function if no BrokerEntryMetadata is found.
+ * The function shouldn't keep a reference to the BrokerEntryMetadata instance after the call completes
+ * since it's a ThreadLocal instance that is reused.
+ *
+ * @param headerAndPayload the header and payload of the message
+ * @param function the function to apply to the BrokerEntryMetadata
+ */
+ public static void peekBrokerEntryMetadataAndConsume(ByteBuf headerAndPayload,
+ Consumer<BrokerEntryMetadata> function) {
+ BrokerEntryMetadata brokerEntryMetadata =
+ parseOrPeekBrokerEntryMetadataIfExist(headerAndPayload, BROKER_ENTRY_METADATA.get(), true);
+ function.accept(brokerEntryMetadata);
}
public static ByteBuf serializeMetadataAndPayload(ChecksumType checksumType,
@@ -1958,28 +2054,28 @@ public class Commands {
return ByteBufPair.get(headers, metadataAndPayload);
}
- public static int getNumberOfMessagesInBatch(ByteBuf metadataAndPayload, String subscription,
- long consumerId) {
- MessageMetadata msgMetadata = peekMessageMetadata(metadataAndPayload, subscription, consumerId);
- if (msgMetadata == null) {
- return -1;
- } else {
- return msgMetadata.getNumMessagesInBatch();
- }
- }
-
public static MessageMetadata peekMessageMetadata(ByteBuf metadataAndPayload, String subscription,
long consumerId) {
+ // save the reader index and restore after parsing
+ int readerIdx = metadataAndPayload.readerIndex();
try {
- // save the reader index and restore after parsing
- int readerIdx = metadataAndPayload.readerIndex();
- MessageMetadata metadata = Commands.parseMessageMetadata(metadataAndPayload);
- metadataAndPayload.readerIndex(readerIdx);
-
+ MessageMetadata metadata = parseMessageMetadata(metadataAndPayload);
return metadata;
} catch (Throwable t) {
log.error("[{}] [{}] Failed to parse message metadata", subscription, consumerId, t);
return null;
+ } finally {
+ metadataAndPayload.readerIndex(readerIdx);
+ }
+ }
+
+ public static void peekMessageMetadata(ByteBuf metadataAndPayload, MessageMetadata msgMetadata) {
+ // save the reader index and restore after parsing
+ int readerIdx = metadataAndPayload.readerIndex();
+ try {
+ parseMessageMetadata(metadataAndPayload, msgMetadata);
+ } finally {
+ metadataAndPayload.readerIndex(readerIdx);
}
}
@@ -1992,26 +2088,28 @@ public class Commands {
*/
public static MessageMetadata peekAndCopyMessageMetadata(
ByteBuf metadataAndPayload, String subscription, long consumerId) {
- final MessageMetadata localMetadata = peekMessageMetadata(metadataAndPayload, subscription, consumerId);
- if (localMetadata == null) {
+ final MessageMetadata metadata = new MessageMetadata();
+ try {
+ peekMessageMetadata(metadataAndPayload, metadata);
+ } catch (Throwable t) {
+ log.error("[{}] [{}] Failed to parse message metadata", subscription, consumerId, t);
return null;
}
- final MessageMetadata metadata = new MessageMetadata();
- metadata.copyFrom(localMetadata);
return metadata;
}
public static final byte[] NONE_KEY = "NONE_KEY".getBytes(StandardCharsets.UTF_8);
public static byte[] peekStickyKey(ByteBuf metadataAndPayload, String topic, String subscription) {
+ int readerIdx = metadataAndPayload.readerIndex();
try {
- int readerIdx = metadataAndPayload.readerIndex();
MessageMetadata metadata = parseMessageMetadata(metadataAndPayload);
- metadataAndPayload.readerIndex(readerIdx);
return resolveStickyKey(metadata);
} catch (Throwable t) {
log.error("[{}] [{}] Failed to peek sticky key from the message metadata", topic, subscription, t);
return NONE_KEY;
+ } finally {
+ metadataAndPayload.readerIndex(readerIdx);
}
}
(pulsar) 02/03: [improve][io] Upgrade to Debezium 3.2.2 (#24712)
Posted by te...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit a2eb1aa42e2c4fa2f72ddc0db7f861556e334e60
Author: Enrique Fernández <en...@bluelabs.eu>
AuthorDate: Mon Sep 8 21:18:31 2025 +0200
[improve][io] Upgrade to Debezium 3.2.2 (#24712)
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
---
pom.xml | 13 +++---
.../apache/pulsar/io/debezium/DebeziumSource.java | 20 +++++----
...tabaseHistory.java => PulsarSchemaHistory.java} | 48 +++++++++++-----------
.../org/apache/pulsar/io/debezium/SerDeUtils.java | 2 +-
...storyTest.java => PulsarSchemaHistoryTest.java} | 40 +++++++++---------
.../io/debezium/mongodb/DebeziumMongoDbSource.java | 7 ++++
.../resources/debezium-mongodb-source-config.yaml | 5 ++-
.../io/debezium/mssql/DebeziumMsSqlSource.java | 7 ++++
.../resources/debezium-mssql-source-config.yaml | 7 ++--
pulsar-io/debezium/mysql/pom.xml | 4 +-
.../io/debezium/mysql/DebeziumMysqlSource.java | 7 ++++
.../resources/debezium-mysql-source-config.yaml | 7 ++--
.../io/debezium/oracle/DebeziumOracleSource.java | 7 ++++
.../resources/debezium-oracle-source-config.yaml | 5 ++-
pulsar-io/debezium/pom.xml | 40 ------------------
.../debezium/postgres/DebeziumPostgresSource.java | 7 ++++
.../resources/debezium-postgres-source-config.yaml | 7 ++--
pulsar-io/mongo/pom.xml | 2 +-
.../apache/pulsar/io/mongodb/MongoSinkTest.java | 19 ++++++++-
.../apache/pulsar/io/mongodb/MongoSourceTest.java | 12 ++++--
src/owasp-dependency-check-suppressions.xml | 43 -------------------
tests/integration/pom.xml | 2 +-
.../containers/DebeziumMongoDbContainer.java | 2 +-
.../containers/DebeziumMySQLContainer.java | 2 +-
.../containers/DebeziumPostgreSqlContainer.java | 2 +-
.../io/sources/PulsarIOSourceRunner.java | 15 +++++++
.../tests/integration/io/sources/SourceTester.java | 2 +
.../debezium/DebeziumMongoDbSourceTester.java | 22 +++++-----
.../debezium/DebeziumMsSqlSourceTester.java | 14 ++++---
.../debezium/DebeziumMySqlSourceTester.java | 7 ++--
.../debezium/DebeziumOracleDbSourceTester.java | 6 ++-
.../debezium/DebeziumPostgreSqlSourceTester.java | 10 ++---
.../debezium/PulsarDebeziumSourcesTest.java | 4 +-
tests/scripts/pre-integ-tests.sh | 4 +-
34 files changed, 202 insertions(+), 199 deletions(-)
diff --git a/pom.xml b/pom.xml
index b397229ea37..8e1cf061bef 100644
--- a/pom.xml
+++ b/pom.xml
@@ -240,19 +240,16 @@ flexible messaging model and an intuitive client API.</description>
<jclouds.version>2.6.0</jclouds.version>
<guice.version>5.1.0</guice.version>
<sqlite-jdbc.version>3.47.1.0</sqlite-jdbc.version>
- <mysql-jdbc.version>8.0.11</mysql-jdbc.version>
- <postgresql-jdbc.version>42.5.5</postgresql-jdbc.version>
+ <postgresql-jdbc.version>42.7.7</postgresql-jdbc.version>
<clickhouse-jdbc.version>0.4.6</clickhouse-jdbc.version>
- <mariadb-jdbc.version>2.7.5</mariadb-jdbc.version>
+ <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>
<elasticsearch-java.version>8.15.3</elasticsearch-java.version>
- <debezium.version>1.9.7.Final</debezium.version>
- <debezium.postgresql.version>42.5.5</debezium.postgresql.version>
- <debezium.mysql.version>8.0.33</debezium.mysql.version>
- <!-- Override version that brings CVE-2022-3143 with debezium -->
- <wildfly-elytron.version>1.15.16.Final</wildfly-elytron.version>
+ <debezium.version>3.2.2.Final</debezium.version>
+ <debezium.postgresql.version>${postgresql-jdbc.version}</debezium.postgresql.version>
+ <debezium.mysql.version>9.4.0</debezium.mysql.version>
<jsonwebtoken.version>0.11.1</jsonwebtoken.version>
<opencensus.version>0.28.0</opencensus.version>
<hadoop3.version>3.4.0</hadoop3.version>
diff --git a/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java b/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java
index 6c422c4f036..f0eaf2938d8 100644
--- a/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java
+++ b/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/DebeziumSource.java
@@ -30,7 +30,7 @@ import org.apache.pulsar.io.kafka.connect.PulsarKafkaWorkerConfig;
@Slf4j
public abstract class DebeziumSource extends KafkaConnectSource {
private static final String DEFAULT_CONVERTER = "org.apache.kafka.connect.json.JsonConverter";
- private static final String DEFAULT_HISTORY = "org.apache.pulsar.io.debezium.PulsarDatabaseHistory";
+ private static final String DEFAULT_HISTORY = "org.apache.pulsar.io.debezium.PulsarSchemaHistory";
private static final String DEFAULT_OFFSET_TOPIC = "debezium-offset-topic";
private static final String DEFAULT_HISTORY_TOPIC = "debezium-history-topic";
@@ -76,6 +76,8 @@ public abstract class DebeziumSource extends KafkaConnectSource {
public abstract void setDbConnectorTask(Map<String, Object> config) throws Exception;
+ public abstract void setDbConnectorClass(Map<String, Object> config) throws Exception;
+
@Override
public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception {
setDbConnectorTask(config);
@@ -87,28 +89,28 @@ public abstract class DebeziumSource extends KafkaConnectSource {
// value.converter
setConfigIfNull(config, PulsarKafkaWorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, DEFAULT_CONVERTER);
- // database.history : implementation class for database history.
- setConfigIfNull(config, HistorizedRelationalDatabaseConnectorConfig.DATABASE_HISTORY.name(), DEFAULT_HISTORY);
+ // schema.history : implementation class for schema history.
+ setConfigIfNull(config, HistorizedRelationalDatabaseConnectorConfig.SCHEMA_HISTORY.name(), DEFAULT_HISTORY);
- // database.history.pulsar.service.url
- String pulsarUrl = (String) config.get(PulsarDatabaseHistory.SERVICE_URL.name());
+ // schema.history.internal.pulsar.service.url
+ String pulsarUrl = (String) config.get(PulsarSchemaHistory.SERVICE_URL.name());
String topicNamespace = topicNamespace(sourceContext);
// topic.namespace
setConfigIfNull(config, PulsarKafkaWorkerConfig.TOPIC_NAMESPACE_CONFIG, topicNamespace);
String sourceName = sourceContext.getSourceName();
- // database.history.pulsar.topic: history topic name
- setConfigIfNull(config, PulsarDatabaseHistory.TOPIC.name(),
+ // schema.history.internal.pulsar.topic: history topic name
+ setConfigIfNull(config, PulsarSchemaHistory.TOPIC.name(),
topicNamespace + "/" + sourceName + "-" + DEFAULT_HISTORY_TOPIC);
// offset.storage.topic: offset topic name
setConfigIfNull(config, PulsarKafkaWorkerConfig.OFFSET_STORAGE_TOPIC_CONFIG,
topicNamespace + "/" + sourceName + "-" + DEFAULT_OFFSET_TOPIC);
- // pass pulsar.client.builder if database.history.pulsar.service.url is not provided
+ // pass pulsar.client.builder if schema.history.internal.pulsar.service.url is not provided
if (StringUtils.isEmpty(pulsarUrl)) {
String pulsarClientBuilder = SerDeUtils.serialize(sourceContext.getPulsarClientBuilder());
- config.put(PulsarDatabaseHistory.CLIENT_BUILDER.name(), pulsarClientBuilder);
+ config.put(PulsarSchemaHistory.CLIENT_BUILDER.name(), pulsarClientBuilder);
}
super.open(config, sourceContext);
diff --git a/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java b/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarSchemaHistory.java
similarity index 86%
rename from pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java
rename to pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarSchemaHistory.java
index 7ca0d309cf9..8ae8c663a8e 100644
--- a/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java
+++ b/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarSchemaHistory.java
@@ -26,12 +26,12 @@ import io.debezium.annotation.ThreadSafe;
import io.debezium.config.Configuration;
import io.debezium.config.Field;
import io.debezium.document.DocumentReader;
-import io.debezium.relational.history.AbstractDatabaseHistory;
-import io.debezium.relational.history.DatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryException;
-import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.AbstractSchemaHistory;
import io.debezium.relational.history.HistoryRecord;
import io.debezium.relational.history.HistoryRecordComparator;
+import io.debezium.relational.history.SchemaHistory;
+import io.debezium.relational.history.SchemaHistoryException;
+import io.debezium.relational.history.SchemaHistoryListener;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
@@ -52,15 +52,15 @@ import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
/**
- * A {@link DatabaseHistory} implementation that records schema changes as normal pulsar messages on the specified
+ * A {@link SchemaHistory} implementation that records schema changes as normal pulsar messages on the specified
* topic, and that recovers the history by establishing a Kafka Consumer re-processing all messages on that topic.
*/
@Slf4j
@ThreadSafe
-public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
+public final class PulsarSchemaHistory extends AbstractSchemaHistory {
public static final Field TOPIC = Field.create(CONFIGURATION_FIELD_PREFIX_STRING + "pulsar.topic")
- .withDisplayName("Database history topic name")
+ .withDisplayName("Schema history topic name")
.withType(Type.STRING)
.withWidth(Width.LONG)
.withImportance(Importance.HIGH)
@@ -97,7 +97,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
TOPIC,
SERVICE_URL,
CLIENT_BUILDER,
- DatabaseHistory.NAME,
+ SchemaHistory.NAME,
READER_CONFIG);
private final ObjectMapper mapper = new ObjectMapper();
@@ -113,7 +113,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
public void configure(
Configuration config,
HistoryRecordComparator comparator,
- DatabaseHistoryListener listener,
+ SchemaHistoryListener listener,
boolean useCatalogBeforeSchema) {
super.configure(config, comparator, listener, useCatalogBeforeSchema);
if (!config.validateAndRecord(ALL_FIELDS, logger::error)) {
@@ -148,9 +148,9 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
}
// Copy the relevant portions of the configuration and add useful defaults ...
- this.dbHistoryName = config.getString(DatabaseHistory.NAME, UUID.randomUUID().toString());
+ this.dbHistoryName = config.getString(SchemaHistory.NAME, UUID.randomUUID().toString());
- log.info("Configure to store the debezium database history {} to pulsar topic {}",
+ log.info("Configure to store the debezium schema history {} to pulsar topic {}",
dbHistoryName, topicName);
}
@@ -163,7 +163,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
p.send("");
} catch (PulsarClientException pce) {
log.error("Failed to initialize storage", pce);
- throw new RuntimeException("Failed to initialize storage", pce);
+ throw new SchemaHistoryException("Failed to initialize storage", pce);
}
}
@@ -172,7 +172,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
try {
pulsarClient = clientBuilder.build();
} catch (PulsarClientException e) {
- throw new RuntimeException("Failed to create pulsar client to pulsar cluster", e);
+ throw new SchemaHistoryException("Failed to create pulsar client to pulsar cluster", e);
}
}
}
@@ -201,18 +201,18 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
}
@Override
- protected void storeRecord(HistoryRecord record) throws DatabaseHistoryException {
+ protected void storeRecord(HistoryRecord record) throws SchemaHistoryException {
if (this.producer == null) {
throw new IllegalStateException("No producer is available. Ensure that 'start()'"
- + " is called before storing database history records.");
+ + " is called before storing schema history records.");
}
if (log.isTraceEnabled()) {
- log.trace("Storing record into database history: {}", record);
+ log.trace("Storing record into schema history: {}", record);
}
try {
producer.send(record.toString());
} catch (PulsarClientException e) {
- throw new DatabaseHistoryException(e);
+ throw new SchemaHistoryException(e);
}
}
@@ -242,7 +242,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
protected void recoverRecords(Consumer<HistoryRecord> records) {
setupClientIfNeeded();
try (Reader<String> historyReader = createHistoryReader()) {
- log.info("Scanning the database history topic '{}'", topicName);
+ log.info("Scanning the schema history topic '{}'", topicName);
// Read all messages in the topic ...
MessageId lastProcessedMessageId = null;
@@ -255,15 +255,15 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
if (!isBlank(msg.getValue())) {
HistoryRecord recordObj = new HistoryRecord(reader.read(msg.getValue()));
if (log.isTraceEnabled()) {
- log.trace("Recovering database history: {}", recordObj);
+ log.trace("Recovering schema history: {}", recordObj);
}
if (!recordObj.isValid()) {
- log.warn("Skipping invalid database history record '{}'. This is often not an issue,"
+ log.warn("Skipping invalid schema history record '{}'. This is often not an issue,"
+ " but if it happens repeatedly please check the '{}' topic.",
recordObj, topicName);
} else {
records.accept(recordObj);
- log.trace("Recovered database history: {}", recordObj);
+ log.trace("Recovered schema history: {}", recordObj);
}
}
lastProcessedMessageId = msg.getMessageId();
@@ -274,7 +274,7 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
throw e;
}
}
- log.info("Successfully completed scanning the database history topic '{}'", topicName);
+ log.info("Successfully completed scanning the schema history topic '{}'", topicName);
} catch (IOException ioe) {
log.error("Encountered issues on recovering history records", ioe);
throw new RuntimeException("Encountered issues on recovering history records", ioe);
@@ -287,8 +287,8 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
try (Reader<String> historyReader = createHistoryReader()) {
return historyReader.hasMessageAvailable();
} catch (IOException e) {
- log.error("Encountered issues on checking existence of database history", e);
- throw new RuntimeException("Encountered issues on checking existence of database history", e);
+ log.error("Encountered issues on checking existence of schema history", e);
+ throw new RuntimeException("Encountered issues on checking existence of schema history", e);
}
}
diff --git a/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/SerDeUtils.java b/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/SerDeUtils.java
index 3d8dd3bd08d..156b19b3f76 100644
--- a/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/SerDeUtils.java
+++ b/pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/SerDeUtils.java
@@ -37,7 +37,7 @@ public class SerDeUtils {
return ois.readObject();
} catch (Exception e) {
throw new RuntimeException(
- "Failed to initialize the pulsar client to store debezium database history", e);
+ "Failed to initialize the pulsar client to store debezium schema history", e);
}
}
diff --git a/pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistoryTest.java b/pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarSchemaHistoryTest.java
similarity index 87%
rename from pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistoryTest.java
rename to pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarSchemaHistoryTest.java
index 4d076ebbc0d..c593b464c4a 100644
--- a/pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistoryTest.java
+++ b/pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarSchemaHistoryTest.java
@@ -26,8 +26,8 @@ import io.debezium.config.Configuration;
import io.debezium.connector.mysql.antlr.MySqlAntlrDdlParser;
import io.debezium.relational.Tables;
import io.debezium.relational.ddl.DdlParser;
-import io.debezium.relational.history.DatabaseHistory;
-import io.debezium.relational.history.DatabaseHistoryListener;
+import io.debezium.relational.history.SchemaHistory;
+import io.debezium.relational.history.SchemaHistoryListener;
import io.debezium.text.ParsingException;
import io.debezium.util.Collect;
import java.io.ByteArrayOutputStream;
@@ -46,11 +46,11 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
/**
- * Test the implementation of {@link PulsarDatabaseHistory}.
+ * Test the implementation of {@link PulsarSchemaHistory}.
*/
-public class PulsarDatabaseHistoryTest extends ProducerConsumerBase {
+public class PulsarSchemaHistoryTest extends ProducerConsumerBase {
- private PulsarDatabaseHistory history;
+ private PulsarSchemaHistory history;
private Map<String, Object> position;
private Map<String, String> source;
private String topicName;
@@ -65,7 +65,7 @@ public class PulsarDatabaseHistoryTest extends ProducerConsumerBase {
source = Collect.hashMapOf("server", "my-server");
setLogPosition(0);
this.topicName = "persistent://my-property/my-ns/schema-changes-topic";
- this.history = new PulsarDatabaseHistory();
+ this.history = new PulsarSchemaHistory();
}
@AfterMethod(alwaysRun = true)
@@ -78,9 +78,9 @@ public class PulsarDatabaseHistoryTest extends ProducerConsumerBase {
private void testHistoryTopicContent(boolean skipUnparseableDDL, boolean testWithClientBuilder,
boolean testWithReaderConfig) throws Exception {
Configuration.Builder configBuidler = Configuration.create()
- .with(PulsarDatabaseHistory.TOPIC, topicName)
- .with(DatabaseHistory.NAME, "my-db-history")
- .with(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, skipUnparseableDDL);
+ .with(PulsarSchemaHistory.TOPIC, topicName)
+ .with(SchemaHistory.NAME, "my-db-history")
+ .with(SchemaHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, skipUnparseableDDL);
if (testWithClientBuilder) {
ClientBuilder builder = PulsarClient.builder().serviceUrl(brokerUrl.toString());
@@ -89,18 +89,18 @@ public class PulsarDatabaseHistoryTest extends ProducerConsumerBase {
oos.writeObject(builder);
oos.flush();
byte[] data = bao.toByteArray();
- configBuidler.with(PulsarDatabaseHistory.CLIENT_BUILDER, Base64.getEncoder().encodeToString(data));
+ configBuidler.with(PulsarSchemaHistory.CLIENT_BUILDER, Base64.getEncoder().encodeToString(data));
}
} else {
- configBuidler.with(PulsarDatabaseHistory.SERVICE_URL, brokerUrl.toString());
+ configBuidler.with(PulsarSchemaHistory.SERVICE_URL, brokerUrl.toString());
}
if (testWithReaderConfig) {
- configBuidler.with(PulsarDatabaseHistory.READER_CONFIG, "{\"subscriptionName\":\"my-subscription\"}");
+ configBuidler.with(PulsarSchemaHistory.READER_CONFIG, "{\"subscriptionName\":\"my-subscription\"}");
}
// Start up the history ...
- history.configure(configBuidler.build(), null, DatabaseHistoryListener.NOOP, true);
+ history.configure(configBuidler.build(), null, SchemaHistoryListener.NOOP, true);
history.start();
// Should be able to call start more than once ...
@@ -159,8 +159,8 @@ public class PulsarDatabaseHistoryTest extends ProducerConsumerBase {
// Stop the history (which should stop the producer) ...
history.stop();
- history = new PulsarDatabaseHistory();
- history.configure(configBuidler.build(), null, DatabaseHistoryListener.NOOP, true);
+ history = new PulsarSchemaHistory();
+ history.configure(configBuidler.build(), null, SchemaHistoryListener.NOOP, true);
// no need to start
// Recover from the very beginning to just past the first change ...
@@ -244,13 +244,13 @@ public class PulsarDatabaseHistoryTest extends ProducerConsumerBase {
// Set history to use dummy topic
Configuration config = Configuration.create()
- .with(PulsarDatabaseHistory.SERVICE_URL, brokerUrl.toString())
- .with(PulsarDatabaseHistory.TOPIC, "persistent://my-property/my-ns/dummytopic")
- .with(DatabaseHistory.NAME, "my-db-history")
- .with(DatabaseHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, true)
+ .with(PulsarSchemaHistory.SERVICE_URL, brokerUrl.toString())
+ .with(PulsarSchemaHistory.TOPIC, "persistent://my-property/my-ns/dummytopic")
+ .with(SchemaHistory.NAME, "my-db-history")
+ .with(SchemaHistory.SKIP_UNPARSEABLE_DDL_STATEMENTS, true)
.build();
- history.configure(config, null, DatabaseHistoryListener.NOOP, true);
+ history.configure(config, null, SchemaHistoryListener.NOOP, true);
history.start();
// dummytopic should not exist yet
diff --git a/pulsar-io/debezium/mongodb/src/main/java/org/apache/pulsar/io/debezium/mongodb/DebeziumMongoDbSource.java b/pulsar-io/debezium/mongodb/src/main/java/org/apache/pulsar/io/debezium/mongodb/DebeziumMongoDbSource.java
index 88acbc61c77..996aa5ac725 100644
--- a/pulsar-io/debezium/mongodb/src/main/java/org/apache/pulsar/io/debezium/mongodb/DebeziumMongoDbSource.java
+++ b/pulsar-io/debezium/mongodb/src/main/java/org/apache/pulsar/io/debezium/mongodb/DebeziumMongoDbSource.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.io.debezium.mongodb;
import java.util.Map;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.pulsar.io.core.SourceContext;
import org.apache.pulsar.io.debezium.DebeziumSource;
@@ -27,8 +28,14 @@ import org.apache.pulsar.io.debezium.DebeziumSource;
* A pulsar source that runs debezium mongodb source.
*/
public class DebeziumMongoDbSource extends DebeziumSource {
+ private static final String DEFAULT_CONNECTOR = "io.debezium.connector.mongodb.MongoDbConnector";
private static final String DEFAULT_TASK = "io.debezium.connector.mongodb.MongoDbConnectorTask";
+ @Override
+ public void setDbConnectorClass(Map<String, Object> config) throws Exception {
+ throwExceptionIfConfigNotMatch(config, ConnectorConfig.CONNECTOR_CLASS_CONFIG, DEFAULT_CONNECTOR);
+ }
+
@Override
public void setDbConnectorTask(Map<String, Object> config) throws Exception {
throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK);
diff --git a/pulsar-io/debezium/mongodb/src/main/resources/debezium-mongodb-source-config.yaml b/pulsar-io/debezium/mongodb/src/main/resources/debezium-mongodb-source-config.yaml
index 8c1564bf4e6..54e0bbd7580 100644
--- a/pulsar-io/debezium/mongodb/src/main/resources/debezium-mongodb-source-config.yaml
+++ b/pulsar-io/debezium/mongodb/src/main/resources/debezium-mongodb-source-config.yaml
@@ -28,10 +28,11 @@ parallelism: 1
configs:
## config for pg, docker image: debezium/example-mongodb:0.8
mongodb.hosts: "rs0/mongodb:27017"
- mongodb.name: "dbserver1"
mongodb.user: "debezium"
mongodb.password: "dbz"
mongodb.task.id: "1"
database.whitelist: "inventory"
+ topic.prefix: "dbserver1"
- database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ schema.history.internal.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ connector.class: "io.debezium.connector.mongodb.MongoDbConnector"
diff --git a/pulsar-io/debezium/mssql/src/main/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSource.java b/pulsar-io/debezium/mssql/src/main/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSource.java
index f60653e7aae..0df5eb6444a 100644
--- a/pulsar-io/debezium/mssql/src/main/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSource.java
+++ b/pulsar-io/debezium/mssql/src/main/java/org/apache/pulsar/io/debezium/mssql/DebeziumMsSqlSource.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.io.debezium.mssql;
import java.util.Map;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.pulsar.io.debezium.DebeziumSource;
@@ -27,8 +28,14 @@ import org.apache.pulsar.io.debezium.DebeziumSource;
* A pulsar source that runs debezium mssql source.
*/
public class DebeziumMsSqlSource extends DebeziumSource {
+ private static final String DEFAULT_CONNECTOR = "io.debezium.connector.sqlserver.SqlServerConnector";
private static final String DEFAULT_TASK = "io.debezium.connector.sqlserver.SqlServerConnectorTask";
+ @Override
+ public void setDbConnectorClass(Map<String, Object> config) throws Exception {
+ throwExceptionIfConfigNotMatch(config, ConnectorConfig.CONNECTOR_CLASS_CONFIG, DEFAULT_CONNECTOR);
+ }
+
@Override
public void setDbConnectorTask(Map<String, Object> config) throws Exception {
throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK);
diff --git a/pulsar-io/debezium/mssql/src/main/resources/debezium-mssql-source-config.yaml b/pulsar-io/debezium/mssql/src/main/resources/debezium-mssql-source-config.yaml
index c99fb48eab0..de09fedcb59 100644
--- a/pulsar-io/debezium/mssql/src/main/resources/debezium-mssql-source-config.yaml
+++ b/pulsar-io/debezium/mssql/src/main/resources/debezium-mssql-source-config.yaml
@@ -30,7 +30,8 @@ configs:
database.port: "1521"
database.user: "sa"
database.password: "MyP@ssword1"
- database.dbname: "MyDB"
- database.server.name: "mssql"
+ database.names: "MyDB"
+ topic.prefix: "mssql"
- database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ schema.history.internal.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ connector.class: "io.debezium.connector.sqlserver.SqlServerConnector"
diff --git a/pulsar-io/debezium/mysql/pom.xml b/pulsar-io/debezium/mysql/pom.xml
index e884b906f8f..b609dec0fe3 100644
--- a/pulsar-io/debezium/mysql/pom.xml
+++ b/pulsar-io/debezium/mysql/pom.xml
@@ -55,8 +55,8 @@
<dependencyManagement>
<dependencies>
<dependency>
- <groupId>mysql</groupId>
- <artifactId>mysql-connector-java</artifactId>
+ <groupId>com.mysql</groupId>
+ <artifactId>mysql-connector-j</artifactId>
<version>${debezium.mysql.version}</version>
</dependency>
</dependencies>
diff --git a/pulsar-io/debezium/mysql/src/main/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSource.java b/pulsar-io/debezium/mysql/src/main/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSource.java
index 633535512d2..7a89fa13193 100644
--- a/pulsar-io/debezium/mysql/src/main/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSource.java
+++ b/pulsar-io/debezium/mysql/src/main/java/org/apache/pulsar/io/debezium/mysql/DebeziumMysqlSource.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.io.debezium.mysql;
import java.util.Map;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.pulsar.io.debezium.DebeziumSource;
@@ -26,8 +27,14 @@ import org.apache.pulsar.io.debezium.DebeziumSource;
* A pulsar source that runs debezium mysql source.
*/
public class DebeziumMysqlSource extends DebeziumSource {
+ private static final String DEFAULT_CONNECTOR = "io.debezium.connector.mysql.MySqlConnector";
private static final String DEFAULT_TASK = "io.debezium.connector.mysql.MySqlConnectorTask";
+ @Override
+ public void setDbConnectorClass(Map<String, Object> config) throws Exception {
+ throwExceptionIfConfigNotMatch(config, ConnectorConfig.CONNECTOR_CLASS_CONFIG, DEFAULT_CONNECTOR);
+ }
+
@Override
public void setDbConnectorTask(Map<String, Object> config) throws Exception {
throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK);
diff --git a/pulsar-io/debezium/mysql/src/main/resources/debezium-mysql-source-config.yaml b/pulsar-io/debezium/mysql/src/main/resources/debezium-mysql-source-config.yaml
index 7a51b0092ac..e278b2276aa 100644
--- a/pulsar-io/debezium/mysql/src/main/resources/debezium-mysql-source-config.yaml
+++ b/pulsar-io/debezium/mysql/src/main/resources/debezium-mysql-source-config.yaml
@@ -32,11 +32,12 @@ configs:
database.user: "debezium"
database.password: "dbz"
database.server.id: "184054"
- database.server.name: "dbserver1"
database.whitelist: "inventory"
+ topic.prefix: "dbserver1"
- database.history.pulsar.topic: "mysql-history-topic"
- database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ schema.history.internal.pulsar.topic: "mysql-history-topic"
+ schema.history.internal.pulsar.service.url: "pulsar://127.0.0.1:6650"
offset.storage.topic: "mysql-offset-topic"
+ connector.class: "io.debezium.connector.mysql.MySqlConnector"
diff --git a/pulsar-io/debezium/oracle/src/main/java/org/apache/pulsar/io/debezium/oracle/DebeziumOracleSource.java b/pulsar-io/debezium/oracle/src/main/java/org/apache/pulsar/io/debezium/oracle/DebeziumOracleSource.java
index cb7363ac8cd..7b17203760b 100644
--- a/pulsar-io/debezium/oracle/src/main/java/org/apache/pulsar/io/debezium/oracle/DebeziumOracleSource.java
+++ b/pulsar-io/debezium/oracle/src/main/java/org/apache/pulsar/io/debezium/oracle/DebeziumOracleSource.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.io.debezium.oracle;
import java.util.Map;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.pulsar.io.debezium.DebeziumSource;
@@ -27,8 +28,14 @@ import org.apache.pulsar.io.debezium.DebeziumSource;
* A pulsar source that runs debezium oracle source.
*/
public class DebeziumOracleSource extends DebeziumSource {
+ private static final String DEFAULT_CONNECTOR = "io.debezium.connector.oracle.OracleConnector";
private static final String DEFAULT_TASK = "io.debezium.connector.oracle.OracleConnectorTask";
+ @Override
+ public void setDbConnectorClass(Map<String, Object> config) throws Exception {
+ throwExceptionIfConfigNotMatch(config, ConnectorConfig.CONNECTOR_CLASS_CONFIG, DEFAULT_CONNECTOR);
+ }
+
@Override
public void setDbConnectorTask(Map<String, Object> config) throws Exception {
throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK);
diff --git a/pulsar-io/debezium/oracle/src/main/resources/debezium-oracle-source-config.yaml b/pulsar-io/debezium/oracle/src/main/resources/debezium-oracle-source-config.yaml
index 94173d68a11..8955585979c 100644
--- a/pulsar-io/debezium/oracle/src/main/resources/debezium-oracle-source-config.yaml
+++ b/pulsar-io/debezium/oracle/src/main/resources/debezium-oracle-source-config.yaml
@@ -33,6 +33,7 @@ configs:
database.user: "sysdba"
database.password: "oracle"
database.dbname: "XE"
- database.server.name: "XE"
+ topic.prefix: "XE"
- database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ schema.history.internal.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ connector.class: "io.debezium.connector.oracle.OracleConnector"
diff --git a/pulsar-io/debezium/pom.xml b/pulsar-io/debezium/pom.xml
index af34475ff3b..8e668d808d2 100644
--- a/pulsar-io/debezium/pom.xml
+++ b/pulsar-io/debezium/pom.xml
@@ -31,46 +31,6 @@
<artifactId>pulsar-io-debezium</artifactId>
<name>Pulsar IO :: Debezium</name>
- <dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>org.wildfly.security</groupId>
- <artifactId>wildfly-elytron-sasl-digest</artifactId>
- <version>${wildfly-elytron.version}</version>
- </dependency>
- <dependency>
- <groupId>org.wildfly.security</groupId>
- <artifactId>wildfly-elytron-sasl-external</artifactId>
- <version>${wildfly-elytron.version}</version>
- </dependency>
- <dependency>
- <groupId>org.wildfly.security</groupId>
- <artifactId>wildfly-elytron-sasl-gs2</artifactId>
- <version>${wildfly-elytron.version}</version>
- </dependency>
- <dependency>
- <groupId>org.wildfly.security</groupId>
- <artifactId>wildfly-elytron-sasl-oauth2</artifactId>
- <version>${wildfly-elytron.version}</version>
- </dependency>
- <dependency>
- <groupId>org.wildfly.security</groupId>
- <artifactId>wildfly-elytron-sasl-plain</artifactId>
- <version>${wildfly-elytron.version}</version>
- </dependency>
- <dependency>
- <groupId>org.wildfly.security</groupId>
- <artifactId>wildfly-elytron-sasl-scram</artifactId>
- <version>${wildfly-elytron.version}</version>
- </dependency>
- <dependency>
- <groupId>org.wildfly.security</groupId>
- <artifactId>wildfly-elytron-password-impl</artifactId>
- <version>${wildfly-elytron.version}</version>
- </dependency>
- </dependencies>
- </dependencyManagement>
-
<modules>
<module>core</module>
<module>mysql</module>
diff --git a/pulsar-io/debezium/postgres/src/main/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSource.java b/pulsar-io/debezium/postgres/src/main/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSource.java
index ef645aad6b1..54ab110edd0 100644
--- a/pulsar-io/debezium/postgres/src/main/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSource.java
+++ b/pulsar-io/debezium/postgres/src/main/java/org/apache/pulsar/io/debezium/postgres/DebeziumPostgresSource.java
@@ -19,6 +19,7 @@
package org.apache.pulsar.io.debezium.postgres;
import java.util.Map;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.TaskConfig;
import org.apache.pulsar.io.debezium.DebeziumSource;
@@ -27,8 +28,14 @@ import org.apache.pulsar.io.debezium.DebeziumSource;
* A pulsar source that runs debezium postgres source.
*/
public class DebeziumPostgresSource extends DebeziumSource {
+ private static final String DEFAULT_CONNECTOR = "io.debezium.connector.postgresql.PostgresConnector";
private static final String DEFAULT_TASK = "io.debezium.connector.postgresql.PostgresConnectorTask";
+ @Override
+ public void setDbConnectorClass(Map<String, Object> config) throws Exception {
+ throwExceptionIfConfigNotMatch(config, ConnectorConfig.CONNECTOR_CLASS_CONFIG, DEFAULT_CONNECTOR);
+ }
+
@Override
public void setDbConnectorTask(Map<String, Object> config) throws Exception {
throwExceptionIfConfigNotMatch(config, TaskConfig.TASK_CLASS_CONFIG, DEFAULT_TASK);
diff --git a/pulsar-io/debezium/postgres/src/main/resources/debezium-postgres-source-config.yaml b/pulsar-io/debezium/postgres/src/main/resources/debezium-postgres-source-config.yaml
index 3f6b7eaaba2..4048c73e53b 100644
--- a/pulsar-io/debezium/postgres/src/main/resources/debezium-postgres-source-config.yaml
+++ b/pulsar-io/debezium/postgres/src/main/resources/debezium-postgres-source-config.yaml
@@ -32,7 +32,8 @@ configs:
database.user: "postgres"
database.password: "postgres"
database.dbname: "postgres"
- database.server.name: "dbserver1"
- schema.whitelist: "inventory"
+ schema.allow.list: "inventory"
+ topic.prefix: "dbserver1"
- database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ schema.history.internal.pulsar.service.url: "pulsar://127.0.0.1:6650"
+ connector.class: "io.debezium.connector.postgresql.PostgresConnector"
diff --git a/pulsar-io/mongo/pom.xml b/pulsar-io/mongo/pom.xml
index ab268f3f841..90d4c88b1c8 100644
--- a/pulsar-io/mongo/pom.xml
+++ b/pulsar-io/mongo/pom.xml
@@ -33,7 +33,7 @@
<name>Pulsar IO :: MongoDB</name>
<properties>
- <mongo-reactivestreams.version>4.1.2</mongo-reactivestreams.version>
+ <mongo-reactivestreams.version>5.2.0</mongo-reactivestreams.version>
</properties>
<dependencies>
diff --git a/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java b/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java
index e442ed0bcad..ea62e546abc 100644
--- a/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java
+++ b/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSinkTest.java
@@ -26,13 +26,22 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.mongodb.MongoBulkWriteException;
+import com.mongodb.ServerAddress;
import com.mongodb.bulk.BulkWriteError;
+import com.mongodb.bulk.BulkWriteInsert;
+import com.mongodb.bulk.BulkWriteResult;
+import com.mongodb.bulk.BulkWriteUpsert;
+import com.mongodb.bulk.WriteConcernError;
+import com.mongodb.internal.bulk.WriteRequest;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.mongodb.reactivestreams.client.MongoDatabase;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.pulsar.functions.api.Record;
import org.apache.pulsar.io.core.SinkContext;
import org.bson.BsonDocument;
@@ -97,7 +106,15 @@ public class MongoSinkTest {
if (throwBulkError) {
List<BulkWriteError> writeErrors = Arrays.asList(
new BulkWriteError(0, "error", new BsonDocument(), 1));
- exc = new MongoBulkWriteException(null, writeErrors, null, null);
+ BulkWriteResult result = BulkWriteResult.acknowledged(
+ WriteRequest.Type.INSERT, 1, 0,
+ Collections.<BulkWriteUpsert>emptyList(),
+ Collections.<BulkWriteInsert>emptyList());
+ WriteConcernError writeConcernError = null;
+ ServerAddress serverAddress = new ServerAddress("localhost", 27017);
+ Set<String> errorLabels = new HashSet<>();
+ exc = new MongoBulkWriteException(result, writeErrors, writeConcernError,
+ serverAddress, errorLabels);
}
subscriber.onError(exc);
return null;
diff --git a/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceTest.java b/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceTest.java
index 6da4b1123fd..7243b365e88 100644
--- a/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceTest.java
+++ b/pulsar-io/mongo/src/test/java/org/apache/pulsar/io/mongodb/MongoSourceTest.java
@@ -110,17 +110,21 @@ public class MongoSourceTest {
source.open(map, mockSourceContext);
- subscriber.onNext(new ChangeStreamDocument<>(
- OperationType.INSERT,
+ subscriber.onNext(new ChangeStreamDocument<Document>(
+ OperationType.INSERT.getValue(),
BsonDocument.parse("{token: true}"),
BsonDocument.parse("{db: \"hello\", coll: \"pulsar\"}"),
BsonDocument.parse("{db: \"hello2\", coll: \"pulsar2\"}"),
new Document("hello", "pulsar"),
+ new Document("hello", "pulsar"), // documentKey
BsonDocument.parse("{_id: 1}"),
new BsonTimestamp(1234, 2),
- null,
+ null, // UpdateDescription
new BsonInt64(1),
- BsonDocument.parse("{id: 1, uid: 1}")));
+ BsonDocument.parse("{id: 1, uid: 1}"),
+ null, // BsonDateTime
+ null, // SplitEvent
+ BsonDocument.parse("{extra: 1}")));
Record<byte[]> record = source.read();
diff --git a/src/owasp-dependency-check-suppressions.xml b/src/owasp-dependency-check-suppressions.xml
index 1ce7392a489..f302c251f96 100644
--- a/src/owasp-dependency-check-suppressions.xml
+++ b/src/owasp-dependency-check-suppressions.xml
@@ -364,49 +364,6 @@
<cpe>cpe:/a:apache:solr</cpe>
</suppress>
- <!-- debezium-related misdetections -->
- <suppress>
- <notes><![CDATA[
- file name: debezium-connector-mysql-1.9.7.Final.jar
- ]]></notes>
- <sha1>74c623b4a9b231e2f0e8f6053b38abd3bc487ce2</sha1>
- <cve>CVE-2017-15945</cve>
- </suppress>
- <suppress>
- <notes><![CDATA[
- file name: mysql-binlog-connector-java-0.27.2.jar
- ]]></notes>
- <sha1>23294cd730e29c00b8ddfbde517dfc955aba090f</sha1>
- <cve>CVE-2017-15945</cve>
- </suppress>
- <suppress>
- <notes><![CDATA[
- file name: debezium-connector-postgres-1.9.7.Final.jar
- ]]></notes>
- <sha1>300ff0bbf795643e914b7c8a6d6ba812a8354d62</sha1>
- <cve>CVE-2015-0241</cve>
- <cve>CVE-2015-0242</cve>
- <cve>CVE-2015-0243</cve>
- <cve>CVE-2015-0244</cve>
- <cve>CVE-2015-3166</cve>
- <cve>CVE-2015-3167</cve>
- <cve>CVE-2016-0766</cve>
- <cve>CVE-2016-0768</cve>
- <cve>CVE-2016-0773</cve>
- <cve>CVE-2016-5423</cve>
- <cve>CVE-2016-5424</cve>
- <cve>CVE-2016-7048</cve>
- <cve>CVE-2017-14798</cve>
- <cve>CVE-2017-7484</cve>
- <cve>CVE-2018-1115</cve>
- <cve>CVE-2019-10127</cve>
- <cve>CVE-2019-10128</cve>
- <cve>CVE-2019-10210</cve>
- <cve>CVE-2019-10211</cve>
- <cve>CVE-2020-25694</cve>
- <cve>CVE-2020-25695</cve>
- <cve>CVE-2021-23214</cve>
- </suppress>
<suppress>
<notes><![CDATA[
file name: protostream-types-4.4.1.Final.jar
diff --git a/tests/integration/pom.xml b/tests/integration/pom.xml
index 1d0ac102f13..031e070a3c5 100644
--- a/tests/integration/pom.xml
+++ b/tests/integration/pom.xml
@@ -34,7 +34,7 @@
<properties>
<integrationTestSuiteFile>pulsar.xml</integrationTestSuiteFile>
- <mongo-reactivestreams.version>4.1.2</mongo-reactivestreams.version>
+ <mongo-reactivestreams.version>5.2.0</mongo-reactivestreams.version>
</properties>
<dependencies>
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMongoDbContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMongoDbContainer.java
index 481725d145b..6fa2e9ff471 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMongoDbContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMongoDbContainer.java
@@ -25,7 +25,7 @@ public class DebeziumMongoDbContainer extends ChaosContainer<DebeziumMongoDbCont
public static final String NAME = "debezium-mongodb-example";
public static final Integer[] PORTS = { 27017 };
- private static final String IMAGE_NAME = "debezium/example-mongodb:0.10";
+ private static final String IMAGE_NAME = "debezium/example-mongodb:3.0.0.Final";
public DebeziumMongoDbContainer(String clusterName) {
super(clusterName, IMAGE_NAME);
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMySQLContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMySQLContainer.java
index 27d624a6c82..cf59cda868d 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMySQLContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumMySQLContainer.java
@@ -26,7 +26,7 @@ public class DebeziumMySQLContainer extends ChaosContainer<DebeziumMySQLContaine
public static final String NAME = "debezium-mysql-example";
static final Integer[] PORTS = { 3306 };
- private static final String IMAGE_NAME = "debezium/example-mysql:0.8";
+ private static final String IMAGE_NAME = "debezium/example-mysql:3.0.0.Final";
public DebeziumMySQLContainer(String clusterName) {
super(clusterName, IMAGE_NAME);
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgreSqlContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgreSqlContainer.java
index 479869c4183..4fd391fd926 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgreSqlContainer.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgreSqlContainer.java
@@ -26,7 +26,7 @@ public class DebeziumPostgreSqlContainer extends ChaosContainer<DebeziumPostgreS
public static final String NAME = "debezium-postgresql-example";
static final Integer[] PORTS = { 5432 };
- private static final String IMAGE_NAME = "debezium/example-postgres:0.10";
+ private static final String IMAGE_NAME = "debezium/example-postgres:3.0.0.Final";
public DebeziumPostgreSqlContainer(String clusterName) {
super(clusterName, IMAGE_NAME);
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/PulsarIOSourceRunner.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/PulsarIOSourceRunner.java
index 92918a14006..7cd1c3ebf66 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/PulsarIOSourceRunner.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/PulsarIOSourceRunner.java
@@ -272,6 +272,21 @@ public class PulsarIOSourceRunner extends PulsarIOTestRunner {
result.getStdout()
);
result.assertNoStderr();
+
+ final String[] packageCommands = {
+ PulsarCluster.ADMIN_SCRIPT,
+ "packages",
+ "delete",
+ "source://" + tenant + "/" + namespace + "/" + sourceName + "@0"
+ };
+
+ try {
+ ContainerExecResult packageResult = pulsarCluster.getAnyWorker().execCmd(packageCommands);
+ log.info("Package metadata deletion result: {}", packageResult.getStdout());
+ } catch (Exception e) {
+ log.warn("Failed to delete package metadata for source://{}/{}/{}@0: {}",
+ tenant, namespace, sourceName, e.getMessage());
+ }
}
protected void getSourceInfoNotFound(String tenant, String namespace, String sourceName) throws Exception {
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java
index 95b075e660a..bd2c95b5772 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/SourceTester.java
@@ -58,6 +58,8 @@ public abstract class SourceTester<ServiceContainerT extends GenericContainer> i
add("source");
add("op");
add("ts_ms");
+ add("ts_us");
+ add("ts_ns");
add("transaction");
}};
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMongoDbSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMongoDbSourceTester.java
index 4c562f3e244..1826d78f351 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMongoDbSourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMongoDbSourceTester.java
@@ -42,15 +42,17 @@ public class DebeziumMongoDbSourceTester extends SourceTester<DebeziumMongoDbCon
this.pulsarCluster = cluster;
pulsarServiceUrl = "pulsar://pulsar-proxy:" + PulsarContainer.BROKER_PORT;
- sourceConfig.put("mongodb.hosts", "rs0/" + DebeziumMongoDbContainer.NAME + ":27017");
- sourceConfig.put("mongodb.name", "dbserver1");
+ sourceConfig.put("mongodb.connection.string",
+ "mongodb://debezium:dbz@" + DebeziumMongoDbContainer.NAME + ":27017/admin?replicaSet=rs0");
sourceConfig.put("mongodb.user", "debezium");
sourceConfig.put("mongodb.password", "dbz");
sourceConfig.put("mongodb.task.id", "1");
+ sourceConfig.put("topic.prefix", "dbserver1");
sourceConfig.put("database.include.list", "inventory");
- sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
+ sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
sourceConfig.put("topic.namespace", "debezium/mongodb");
- sourceConfig.put("capture.mode", "oplog");
+ sourceConfig.put("capture.mode", "change_streams_update_full");
+ sourceConfig.put("connector.class", "io.debezium.connector.mongodb.MongoDbConnector");
}
@Override
@@ -69,10 +71,10 @@ public class DebeziumMongoDbSourceTester extends SourceTester<DebeziumMongoDbCon
@Override
public void prepareInsertEvent() throws Exception {
this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
- "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "--eval 'db.products.find()'");
this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
- "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "--eval 'db.products.insert({ "
+ "_id : NumberLong(\"110\"),"
+ "name : \"test-debezium\","
@@ -84,20 +86,20 @@ public class DebeziumMongoDbSourceTester extends SourceTester<DebeziumMongoDbCon
@Override
public void prepareDeleteEvent() throws Exception {
this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
- "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "--eval 'db.products.find()'");
this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
- "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "--eval 'db.products.deleteOne({name : \"test-debezium-update\"})'");
}
@Override
public void prepareUpdateEvent() throws Exception {
this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
- "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "--eval 'db.products.find()'");
this.debeziumMongoDbContainer.execCmd("/bin/bash", "-c",
- "mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory "
+ "--eval 'db.products.update({"
+ "_id : 110},"
+ "{$set:{name:\"test-debezium-update\", description: \"this is update description\"}})'");
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMsSqlSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMsSqlSourceTester.java
index 25a7544f52b..2295a670660 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMsSqlSourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMsSqlSourceTester.java
@@ -58,11 +58,15 @@ public class DebeziumMsSqlSourceTester extends SourceTester<DebeziumMsSqlContain
sourceConfig.put("database.port", "1433");
sourceConfig.put("database.user", "sa");
sourceConfig.put("database.password", DebeziumMsSqlContainer.SA_PASSWORD);
- sourceConfig.put("database.server.name", "mssql");
- sourceConfig.put("database.dbname", "TestDB");
+ sourceConfig.put("database.names", "TestDB");
+ sourceConfig.put("database.encrypt", "false");
sourceConfig.put("snapshot.mode", "schema_only");
- sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
+ sourceConfig.put("schema.history.internal.pulsar.topic", "debezium-schema-history-mssql");
+ sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
+ sourceConfig.put("topic.prefix", "mssql");
sourceConfig.put("topic.namespace", "debezium/mssql");
+ sourceConfig.put("task.id", "1");
+ sourceConfig.put("connector.class", "io.debezium.connector.sqlserver.SqlServerConnector");
}
@Override
@@ -145,12 +149,12 @@ public class DebeziumMsSqlSourceTester extends SourceTester<DebeziumMsSqlContain
@Override
public String keyContains() {
- return "mssql.dbo.customers.Key";
+ return "TestDB";
}
@Override
public String valueContains() {
- return "mssql.dbo.customers.Value";
+ return "TestDB";
}
@Override
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMySqlSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMySqlSourceTester.java
index 2c457cd2fb9..7f9ccbf5d20 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMySqlSourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumMySqlSourceTester.java
@@ -58,15 +58,16 @@ public class DebeziumMySqlSourceTester extends SourceTester<DebeziumMySQLContain
sourceConfig.put("database.user", "debezium");
sourceConfig.put("database.password", "dbz");
sourceConfig.put("database.server.id", "184054");
- sourceConfig.put("database.server.name", "dbserver1");
- sourceConfig.put("database.whitelist", "inventory");
+ sourceConfig.put("topic.prefix", "dbserver1");
+ sourceConfig.put("database.include.list", "inventory");
if (!testWithClientBuilder) {
- sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
+ sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
}
sourceConfig.put("key.converter", converterClassName);
sourceConfig.put("value.converter", converterClassName);
sourceConfig.put("topic.namespace", "debezium/mysql-"
+ (converterClassName.endsWith("AvroConverter") ? "avro" : "json"));
+ sourceConfig.put("connector.class", "io.debezium.connector.mysql.MySqlConnector");
}
@Override
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumOracleDbSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumOracleDbSourceTester.java
index 41db0a7cc18..39b4361ee8a 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumOracleDbSourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumOracleDbSourceTester.java
@@ -58,13 +58,15 @@ public class DebeziumOracleDbSourceTester extends SourceTester<DebeziumOracleDbC
sourceConfig.put("database.port", "1521");
sourceConfig.put("database.user", "dbzuser");
sourceConfig.put("database.password", "dbz");
- sourceConfig.put("database.server.name", "XE");
+ sourceConfig.put("topic.prefix", "XE");
sourceConfig.put("database.dbname", "XE");
sourceConfig.put("snapshot.mode", "schema_only");
sourceConfig.put("schema.include.list", "inv");
- sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
+ sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
sourceConfig.put("topic.namespace", "debezium/oracle");
+
+ sourceConfig.put("connector.class", "io.debezium.connector.oracle.OracleConnector");
}
@Override
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumPostgreSqlSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumPostgreSqlSourceTester.java
index 75b071c0bd8..e23a5f1ad50 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumPostgreSqlSourceTester.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/DebeziumPostgreSqlSourceTester.java
@@ -70,13 +70,13 @@ public class DebeziumPostgreSqlSourceTester extends SourceTester<DebeziumPostgre
sourceConfig.put("database.port", "5432");
sourceConfig.put("database.user", "postgres");
sourceConfig.put("database.password", "postgres");
- sourceConfig.put("database.server.id", "184055");
- sourceConfig.put("database.server.name", "dbserver1");
+ sourceConfig.put("topic.prefix", "dbserver1");
sourceConfig.put("database.dbname", "postgres");
- sourceConfig.put("schema.whitelist", "inventory");
- sourceConfig.put("table.blacklist", "inventory.spatial_ref_sys,inventory.geom");
- sourceConfig.put("database.history.pulsar.service.url", pulsarServiceUrl);
+ sourceConfig.put("schema.include.list", "inventory");
+ sourceConfig.put("table.exclude.list", "inventory.spatial_ref_sys,inventory.geom");
+ sourceConfig.put("schema.history.internal.pulsar.service.url", pulsarServiceUrl);
sourceConfig.put("topic.namespace", "debezium/postgresql");
+ sourceConfig.put("connector.class", "io.debezium.connector.postgresql.PostgresConnector");
}
@Override
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java
index 50160d94eef..eb229410391 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/sources/debezium/PulsarDebeziumSourcesTest.java
@@ -91,7 +91,7 @@ public class PulsarDebeziumSourcesTest extends PulsarIOTestBase {
+ "-" + functionRuntimeType + "-" + randomName(8);
// This is the binlog count that contained in mysql container.
- final int numMessages = 47;
+ final int numMessages = 52;
@Cleanup
PulsarClient client = PulsarClient.builder()
@@ -214,7 +214,7 @@ public class PulsarDebeziumSourcesTest extends PulsarIOTestBase {
final String tenant = TopicName.PUBLIC_TENANT;
final String namespace = TopicName.DEFAULT_NAMESPACE;
final String outputTopicName = "debe-output-topic-name-" + testId.getAndIncrement();
- final String consumeTopicName = "debezium/mssql/mssql.dbo.customers";
+ final String consumeTopicName = "debezium/mssql/mssql.TestDB.dbo.customers";
final String sourceName = "test-source-debezium-mssql-" + functionRuntimeType + "-" + randomName(8);
final int numMessages = 1;
diff --git a/tests/scripts/pre-integ-tests.sh b/tests/scripts/pre-integ-tests.sh
index 9f564a3f5f9..f8ba1bdc701 100755
--- a/tests/scripts/pre-integ-tests.sh
+++ b/tests/scripts/pre-integ-tests.sh
@@ -30,5 +30,5 @@ docker pull apachepulsar/s3mock:latest
docker pull alpine/socat:latest
docker pull cassandra:3
docker pull confluentinc/cp-kafka:4.0.0
-docker pull debezium/example-mysql:0.8
-docker pull mysql:5.7.22
+docker pull debezium/example-mysql:3.0.0.Final
+docker pull mysql:9.1.0