This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 651ff41a3a9 Upgrade Pulsar to 5.0.0 (#19787)
651ff41a3a9 is described below
commit 651ff41a3a94b2669386f3e530ad6da73ac585ef
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Oct 7 16:38:28 2026 -0700
Upgrade Pulsar to 5.0.0 (#19787)
---
.../pinot-stream-ingestion/pinot-pulsar/pom.xml | 17 ++-
.../plugin/stream/pulsar/PulsarMessageId.java | 138 +++++++++++++++++++++
.../PulsarPartitionLevelConnectionHandler.java | 32 +++--
.../pulsar/PulsarPartitionLevelConsumer.java | 3 +-
.../pinot/plugin/stream/pulsar/PulsarUtils.java | 10 +-
.../plugin/stream/pulsar/PulsarConsumerTest.java | 6 +-
.../stream/pulsar/PulsarMessageBatchTest.java | 5 +
.../plugin/stream/pulsar/PulsarMessageIdTest.java | 64 ++++++++++
.../plugin/stream/pulsar/PulsarUtilsTest.java | 17 +++
pom.xml | 27 ++--
10 files changed, 280 insertions(+), 39 deletions(-)
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/pom.xml
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/pom.xml
index a8f7999e5da..239fa283465 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/pom.xml
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/pom.xml
@@ -39,14 +39,21 @@
<dependencies>
<dependency>
<groupId>org.apache.pulsar</groupId>
- <artifactId>pulsar-client</artifactId>
+ <artifactId>pulsar-client-all</artifactId>
</dependency>
-
+ <!-- Pulsar declares these with runtime scope, which takes precedence over
the provided scope inherited through
+ pinot-spi and removes them from the compile classpath -->
<dependency>
- <groupId>org.apache.pulsar</groupId>
- <artifactId>pulsar-client-admin</artifactId>
- <scope>test</scope>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.jspecify</groupId>
+ <artifactId>jspecify</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers-pulsar</artifactId>
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageId.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageId.java
new file mode 100644
index 00000000000..f9b315f5e12
--- /dev/null
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageId.java
@@ -0,0 +1,138 @@
+/**
+ * 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.pinot.plugin.stream.pulsar;
+
+import java.util.Arrays;
+import org.apache.pulsar.client.api.MessageIdAdv;
+
+
+/// [MessageIdAdv] built from its components, for message ids computed by
Pinot (e.g. the id of the next message in a
+/// partition). The Pulsar client keeps its message id implementations
internal, and offers no public factory for them.
+///
+/// Equality, hash code, ordering and string form follow the built-in Pulsar
message ids, so instances can be compared
+/// with the message ids returned by the client. [#toByteArray] produces the
same `MessageIdData` protobuf bytes as the
+/// built-in message ids, which can be deserialized with
`MessageId.fromByteArray()`. The ack set used to acknowledge
+/// batched messages is not tracked, as these message ids are not acknowledged.
+///
+/// This class is immutable and thread-safe.
+final class PulsarMessageId implements MessageIdAdv {
+ // Protobuf tags (field number << 3 | wire type VARINT) of the
`MessageIdData` fields
+ private static final byte LEDGER_ID_TAG = 1 << 3;
+ private static final byte ENTRY_ID_TAG = 2 << 3;
+ private static final byte PARTITION_TAG = 3 << 3;
+ private static final byte BATCH_INDEX_TAG = 4 << 3;
+ private static final byte BATCH_SIZE_TAG = 6 << 3;
+ // Up to 5 fields, each with a 1-byte tag and a varint of at most 10 bytes
+ private static final int MAX_SERIALIZED_SIZE = 55;
+
+ private final long _ledgerId;
+ private final long _entryId;
+ private final int _partitionIndex;
+ private final int _batchIndex;
+ private final int _batchSize;
+
+ /// Creates the message id of a non-batched message.
+ PulsarMessageId(long ledgerId, long entryId, int partitionIndex) {
+ this(ledgerId, entryId, partitionIndex, -1, 0);
+ }
+
+ /// Creates the message id of a message at `batchIndex` in a batch of
`batchSize` messages.
+ PulsarMessageId(long ledgerId, long entryId, int partitionIndex, int
batchIndex, int batchSize) {
+ _ledgerId = ledgerId;
+ _entryId = entryId;
+ _partitionIndex = partitionIndex;
+ _batchIndex = batchIndex;
+ _batchSize = batchSize;
+ }
+
+ @Override
+ public long getLedgerId() {
+ return _ledgerId;
+ }
+
+ @Override
+ public long getEntryId() {
+ return _entryId;
+ }
+
+ @Override
+ public int getPartitionIndex() {
+ return _partitionIndex;
+ }
+
+ @Override
+ public int getBatchIndex() {
+ return _batchIndex;
+ }
+
+ @Override
+ public int getBatchSize() {
+ return _batchSize;
+ }
+
+ @Override
+ public byte[] toByteArray() {
+ byte[] buffer = new byte[MAX_SERIALIZED_SIZE];
+ int size = writeField(buffer, 0, LEDGER_ID_TAG, _ledgerId);
+ size = writeField(buffer, size, ENTRY_ID_TAG, _entryId);
+ if (_partitionIndex >= 0) {
+ size = writeField(buffer, size, PARTITION_TAG, _partitionIndex);
+ }
+ if (_batchIndex != -1) {
+ size = writeField(buffer, size, BATCH_INDEX_TAG, _batchIndex);
+ }
+ size = writeField(buffer, size, BATCH_SIZE_TAG, _batchSize);
+ return Arrays.copyOf(buffer, size);
+ }
+
+ /// Writes the tag and the varint encoded value of a field at `offset`, and
returns the offset after it. `int32`
+ /// values are sign-extended to 64 bits like protobuf does, so negative
values take 10 bytes.
+ private static int writeField(byte[] buffer, int offset, byte tag, long
value) {
+ buffer[offset++] = tag;
+ while ((value & ~0x7FL) != 0) {
+ buffer[offset++] = (byte) ((value & 0x7F) | 0x80);
+ value >>>= 7;
+ }
+ buffer[offset++] = (byte) value;
+ return offset;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof MessageIdAdv that)) {
+ return false;
+ }
+ return _ledgerId == that.getLedgerId() && _entryId == that.getEntryId()
+ && _partitionIndex == that.getPartitionIndex() && _batchIndex ==
that.getBatchIndex();
+ }
+
+ @Override
+ public int hashCode() {
+ return (int) (31 * (_ledgerId + 31 * _entryId) + (31 * (long)
_partitionIndex) + _batchIndex);
+ }
+
+ @Override
+ public String toString() {
+ String str = _ledgerId + ":" + _entryId + ":" + _partitionIndex;
+ return _batchIndex != -1 ? str + ":" + _batchIndex : str;
+ }
+}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
index 29b3175deec..444b368ba8e 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConnectionHandler.java
@@ -18,26 +18,28 @@
*/
package org.apache.pinot.plugin.stream.pulsar;
+import com.google.common.base.Preconditions;
import java.io.Closeable;
import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
+import java.util.Map;
import java.util.Optional;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.spi.stream.StreamConfig;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminBuilder;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.ClientBuilder;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.impl.auth.oauth2.AuthenticationFactoryOAuth2;
-
-import static com.google.common.base.Preconditions.checkArgument;
/// Manages the Pulsar client connection, given the partition id and
[PulsarConfig]
public class PulsarPartitionLevelConnectionHandler implements Closeable {
+ // Built-in OAuth2 authentication plugin, which the Pulsar client resolves
by this name even when it is shaded
+ private static final String OAUTH2_AUTH_PLUGIN_CLASS_NAME =
+ "org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2";
+
protected final PulsarConfig _config;
protected final String _clientId;
protected final PulsarClient _pulsarClient;
@@ -63,7 +65,7 @@ public class PulsarPartitionLevelConnectionHandler implements
Closeable {
}
protected PulsarAdmin createPulsarAdmin() {
- checkArgument(StringUtils.isNotBlank(_config.getServiceHttpUrl()),
+
Preconditions.checkArgument(StringUtils.isNotBlank(_config.getServiceHttpUrl()),
"Service HTTP URL must be provided to perform admin operations");
PulsarAdminBuilder adminBuilder =
PulsarAdmin.builder().serviceHttpUrl(_config.getServiceHttpUrl());
@@ -82,7 +84,7 @@ public class PulsarPartitionLevelConnectionHandler implements
Closeable {
///
/// @return an Authentication object
private Authentication authenticationConfig()
- throws MalformedURLException {
+ throws IOException {
String authenticationToken = _config.getAuthenticationToken();
if (StringUtils.isNotBlank(authenticationToken)) {
return AuthenticationFactory.token(authenticationToken);
@@ -95,15 +97,21 @@ public class PulsarPartitionLevelConnectionHandler
implements Closeable {
///
/// @return an OAuth2 Authentication object
private Authentication oAuth2AuthenticationConfig()
- throws MalformedURLException {
+ throws IOException {
String issuerUrl = _config.getIssuerUrl();
String credentialsFilePath = _config.getCredentialsFilePath();
String audience = _config.getAudience();
- if (StringUtils.isNotBlank(issuerUrl) &&
StringUtils.isNotBlank(credentialsFilePath) && StringUtils.isNotBlank(
- audience)) {
- return AuthenticationFactoryOAuth2.clientCredentials(new URL(issuerUrl),
new URL(credentialsFilePath),
- audience);
+ if (StringUtils.isNotBlank(issuerUrl)
+ && StringUtils.isNotBlank(credentialsFilePath)
+ && StringUtils.isNotBlank(audience)) {
+ Map<String, String> authParams = Map.of(
+ "type", "client_credentials",
+ "issuerUrl", issuerUrl,
+ "privateKey", credentialsFilePath,
+ "audience", audience
+ );
+ return AuthenticationFactory.create(OAUTH2_AUTH_PLUGIN_CLASS_NAME,
JsonUtils.objectToString(authParams));
}
return null;
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java
index fb24bda7b1b..32adb6cd208 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarPartitionLevelConsumer.java
@@ -29,7 +29,6 @@ import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Reader;
-import org.apache.pulsar.common.naming.TopicName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,7 +44,7 @@ public class PulsarPartitionLevelConsumer extends
PulsarPartitionLevelConnection
super(clientId, streamConfig);
String topicName = _config.getPulsarTopicName();
try {
- String partitionName = TopicName.getTopicPartitionNameString(topicName,
partition);
+ String partitionName = topicName + "-partition-" + partition;
_reader =
_pulsarClient.newReader().topic(partitionName).startMessageId(MessageId.earliest)
.startMessageIdInclusive().create();
} catch (Exception e) {
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtils.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtils.java
index d9c00045288..a2ff07c2b6e 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtils.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/main/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtils.java
@@ -22,7 +22,6 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import java.nio.ByteBuffer;
import java.util.Base64;
-import java.util.BitSet;
import java.util.Map;
import java.util.Set;
import org.apache.commons.lang3.StringUtils;
@@ -34,8 +33,6 @@ import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
-import org.apache.pulsar.client.impl.BatchMessageIdImpl;
-import org.apache.pulsar.client.impl.MessageIdImpl;
public class PulsarUtils {
@@ -121,14 +118,13 @@ public class PulsarUtils {
int batchSize = messageIdAdv.getBatchSize();
if (batchSize > 0) {
int batchIndex = messageIdAdv.getBatchIndex();
- BitSet ackSet = messageIdAdv.getAckSet();
if (batchIndex < batchSize - 1) {
- return new BatchMessageIdImpl(ledgerId, entryId, partitionIndex,
batchIndex + 1, batchSize, ackSet);
+ return new PulsarMessageId(ledgerId, entryId, partitionIndex,
batchIndex + 1, batchSize);
} else {
- return new BatchMessageIdImpl(ledgerId, entryId + 1, partitionIndex,
0, batchSize, ackSet);
+ return new PulsarMessageId(ledgerId, entryId + 1, partitionIndex, 0,
batchSize);
}
} else {
- return new MessageIdImpl(ledgerId, entryId + 1, partitionIndex);
+ return new PulsarMessageId(ledgerId, entryId + 1, partitionIndex);
}
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java
index 42c91edcd07..b4ca696d526 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarConsumerTest.java
@@ -37,12 +37,12 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.Topics;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TopicMetadata;
-import org.apache.pulsar.client.impl.BatchMessageIdImpl;
import org.testcontainers.pulsar.PulsarContainer;
import org.testcontainers.utility.DockerImageName;
import org.testng.annotations.AfterClass;
@@ -120,7 +120,7 @@ public class PulsarConsumerTest {
producer.flush();
for (int i = 0; i < NUM_RECORDS_PER_PARTITION; i++) {
MessageId messageId = futures.get(i).get();
- assertFalse(messageId instanceof BatchMessageIdImpl);
+ assertEquals(((MessageIdAdv) messageId).getBatchIndex(), -1);
messageIds.add(messageId);
}
}
@@ -147,7 +147,7 @@ public class PulsarConsumerTest {
producer.flush();
for (int i = 0; i < NUM_RECORDS_PER_PARTITION; i++) {
MessageId messageId = futures.get(i).get();
- assertTrue(messageId instanceof BatchMessageIdImpl);
+ assertTrue(((MessageIdAdv) messageId).getBatchIndex() >= 0);
messageIds.add(messageId);
}
}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageBatchTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageBatchTest.java
index 70f8ffa3e98..512e725aa69 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageBatchTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageBatchTest.java
@@ -160,6 +160,11 @@ public class PulsarMessageBatchTest {
return new byte[0];
}
+ @Override
+ public Optional<byte[]> getSchemaId() {
+ return Optional.empty();
+ }
+
@Override
public boolean isReplicated() {
return false;
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageIdTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageIdTest.java
new file mode 100644
index 00000000000..70ae16ad845
--- /dev/null
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarMessageIdTest.java
@@ -0,0 +1,64 @@
+/**
+ * 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.pinot.plugin.stream.pulsar;
+
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageIdAdv;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+public class PulsarMessageIdTest {
+
+ @Test
+ public void testNonBatchedMessageId()
+ throws Exception {
+ testAgainstBuiltInMessageId(new PulsarMessageId(1L << 40, 12345, 3));
+ }
+
+ @Test
+ public void testNonPartitionedMessageId()
+ throws Exception {
+ testAgainstBuiltInMessageId(new PulsarMessageId(1L << 40, 12345, -1));
+ }
+
+ @Test
+ public void testBatchedMessageId()
+ throws Exception {
+ testAgainstBuiltInMessageId(new PulsarMessageId(1L << 40, 12345, 3, 7,
10));
+ }
+
+ /// Deserializes the message id into the built-in Pulsar message id, and
checks that the two are interchangeable.
+ private static void testAgainstBuiltInMessageId(PulsarMessageId messageId)
+ throws Exception {
+ byte[] bytes = messageId.toByteArray();
+ MessageIdAdv builtInMessageId = (MessageIdAdv)
MessageId.fromByteArray(bytes);
+ assertEquals(builtInMessageId.getLedgerId(), messageId.getLedgerId());
+ assertEquals(builtInMessageId.getEntryId(), messageId.getEntryId());
+ assertEquals(builtInMessageId.getPartitionIndex(),
messageId.getPartitionIndex());
+ assertEquals(builtInMessageId.getBatchIndex(), messageId.getBatchIndex());
+ assertEquals(builtInMessageId.getBatchSize(), messageId.getBatchSize());
+ assertEquals(builtInMessageId.toByteArray(), bytes);
+ assertEquals(builtInMessageId, messageId);
+ assertEquals(builtInMessageId.hashCode(), messageId.hashCode());
+ assertEquals(builtInMessageId.compareTo(messageId), 0);
+ assertEquals(builtInMessageId.toString(), messageId.toString());
+ }
+}
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtilsTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtilsTest.java
index be1c5050c8f..3afd447d42f 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtilsTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-pulsar/src/test/java/org/apache/pinot/plugin/stream/pulsar/PulsarUtilsTest.java
@@ -92,4 +92,21 @@ public class PulsarUtilsTest {
assertEquals(streamMessage.getKey(), key.getBytes(StandardCharsets.UTF_8));
assertEquals(streamMessage.getValue(), stitchedValueBytes);
}
+
+ @Test
+ public void testGetNextMessageId()
+ throws Exception {
+ // Non-batched message
+ testGetNextMessageId(new PulsarMessageId(5, 10, 1), new PulsarMessageId(5,
11, 1));
+ // Batched message in the middle of the batch
+ testGetNextMessageId(new PulsarMessageId(5, 10, 1, 3, 10), new
PulsarMessageId(5, 10, 1, 4, 10));
+ // Last message of the batch
+ testGetNextMessageId(new PulsarMessageId(5, 10, 1, 9, 10), new
PulsarMessageId(5, 11, 1, 0, 10));
+ }
+
+ private static void testGetNextMessageId(PulsarMessageId messageId,
PulsarMessageId expectedNextMessageId)
+ throws Exception {
+ MessageId builtInMessageId =
MessageId.fromByteArray(messageId.toByteArray());
+ assertEquals(PulsarUtils.getNextMessageId(builtInMessageId).toByteArray(),
expectedNextMessageId.toByteArray());
+ }
}
diff --git a/pom.xml b/pom.xml
index 3f8643f2ac9..9b353609547 100644
--- a/pom.xml
+++ b/pom.xml
@@ -191,6 +191,7 @@
<helix.version>2.0.1</helix.version>
<zkclient.version>0.11</zkclient.version>
<jackson.version>2.22.3</jackson.version>
+ <opentelemetry.version>1.66.0</opentelemetry.version>
<fory.version>1.7.6</fory.version>
<zookeeper.version>3.9.6</zookeeper.version>
<async-http-client.version>3.0.13</async-http-client.version>
@@ -261,7 +262,7 @@
<kafka3.version>3.9.2</kafka3.version>
<kafka4.version>4.2.0</kafka4.version>
<confluent.version>8.2.0</confluent.version>
- <pulsar.version>4.0.13</pulsar.version>
+ <pulsar.version>5.0.0</pulsar.version>
<flink.version>2.3.0</flink.version>
<!-- Apache Commons Libraries -->
@@ -973,6 +974,17 @@
<scope>import</scope>
</dependency>
+ <!-- OpenTelemetry -->
+ <!-- opentelemetry-bom aligns the OpenTelemetry API and SDK, which
Google libraries and Pulsar request at
+ different versions -->
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-bom</artifactId>
+ <version>${opentelemetry.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+
<!-- Apache Commons Libraries -->
<dependency>
<groupId>org.apache.commons</groupId>
@@ -1162,7 +1174,8 @@
<type>pom</type>
<scope>import</scope>
</dependency>
- <!-- libraries-bom should be imported after protobuf-bom, grpc-bom and
guava-bom to honor their versions -->
+ <!-- libraries-bom should be imported after protobuf-bom, grpc-bom,
guava-bom and opentelemetry-bom to honor their
+ versions -->
<dependency>
<groupId>com.google.cloud</groupId>
<artifactId>libraries-bom</artifactId>
@@ -1531,12 +1544,6 @@
<artifactId>swagger-jersey2-jaxrs</artifactId>
<version>${swagger.version}</version>
</dependency>
- <!-- Also used by pulsar-common -->
- <dependency>
- <groupId>io.swagger</groupId>
- <artifactId>swagger-annotations</artifactId>
- <version>${swagger.version}</version>
- </dependency>
<dependency>
<groupId>org.webjars</groupId>
<artifactId>swagger-ui</artifactId>
@@ -1841,7 +1848,7 @@
<artifactId>dnsjava</artifactId>
<version>${dnsjava.version}</version>
</dependency>
- <!-- Eclipse jetty dependencies in Hadoop/Spark/Pulsar -->
+ <!-- Eclipse jetty dependencies in Hadoop/Spark -->
<dependency>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-client</artifactId>
@@ -1912,7 +1919,7 @@
<artifactId>bcutil-jdk18on</artifactId>
<version>${bouncycastle.version}</version>
</dependency>
- <!-- Used by ORC, Parquet and Pulsar -->
+ <!-- Used by ORC and Parquet -->
<dependency>
<groupId>io.airlift</groupId>
<artifactId>aircompressor</artifactId>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
