适配小红书的Pulsar客户端版本

Author: xiangying

pom.xml

@@ -29,14 +29,27 @@
     <packaging>pom</packaging>
     <name>Pulsar Java Contrib</name>
 
+    <distributionManagement>
+        <snapshotRepository>
+            <id>snapshots</id>
+            <name>maven snapshot repository</name>
+            <url>https://artifactory.devops.xiaohongshu.com/artifactory/maven-snapshots/</url>
+        </snapshotRepository>
+        <repository>
+            <id>releases</id>
+            <name>maven releases repository</name>
+            <url>https://artifactory.devops.xiaohongshu.com/artifactory/maven-releases/</url>
+        </repository>
+    </distributionManagement>
+
     <properties>
         <lombok.version>1.18.32</lombok.version>
         <slf4j.version>2.0.13</slf4j.version>
         <maven.compiler.release>17</maven.compiler.release>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <pulsar.version>3.3.1</pulsar.version>
-        <maven-checkstyle-plugin.version>3.4.0</maven-checkstyle-plugin.version>
+        <pulsar.version>3.0.5-r0.0.5</pulsar.version>
+        <maven-checkstyle-plugin.version>3.3.0</maven-checkstyle-plugin.version>
         <puppycrawl.checkstyle.version>8.45.1</puppycrawl.checkstyle.version>
         <spotless-maven-plugin.version>2.43.0</spotless-maven-plugin.version>
         <maven-compiler-plugin>3.13.0</maven-compiler-plugin>
@@ -75,13 +88,6 @@
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
-            <dependency>
-                <groupId>org.apache.pulsar</groupId>
-                <artifactId>pulsar-bom</artifactId>
-                <version>${pulsar.version}</version>
-                <type>pom</type>
-                <scope>import</scope>
-            </dependency>
             <dependency>
                 <groupId>org.apache.commons</groupId>
                 <artifactId>commons-pool2</artifactId>

pulsar-client-common-contrib/pom.xml

@@ -23,13 +23,16 @@
         <version>1.0.0-SNAPSHOT</version>
     </parent>
     <inceptionYear>2024</inceptionYear>
+    <properties>
+        <maven.compiler.release>8</maven.compiler.release>
+    </properties>
 
     <artifactId>pulsar-client-common-contrib</artifactId>
 
     <dependencies>
         <dependency>
             <groupId>org.apache.pulsar</groupId>
-            <artifactId>pulsar-client-all</artifactId>
+            <artifactId>pulsar-client</artifactId>
             <version>${pulsar.version}</version>
         </dependency>
         <dependency>

pulsar-client-common-contrib/src/main/java/org/apache/pulsar/client/util/PulsarAdminUtils.java

@@ -79,20 +79,23 @@ private static long extractMessageIndex(String topic, MessageIdAdv messageId, St
     }
 
     private static long getMessageIndex(String topic, MessageIdAdv messageId, String brokerCluster,
-                                         PulsarAdmin pulsarAdmin) throws PulsarAdminException {
-        List<Message<byte[]>> messages =
-                pulsarAdmin.topics().getMessagesById(topic, messageId.getLedgerId(), messageId.getEntryId());
+                                        PulsarAdmin pulsarAdmin) throws PulsarAdminException {
+        Message<byte[]> message =
+                pulsarAdmin.topics().getMessageById(topic, messageId.getLedgerId(), messageId.getEntryId());
 
-        if (messages == null || messages.isEmpty()) {
-            throw new PulsarAdminException("No messages found for " + messageId+ " in topic " + topic);
+        if (message == null) {
+            throw new PulsarAdminException("No messages found for " + messageId + " in topic " + topic);
         }
 
-        return messages.stream().map(Message::getIndex).filter(Optional::isPresent).mapToLong(opt -> {
-            long index = opt.get();
+        Optional<Long> indexOptional = message.getIndex();
+        if (indexOptional.isPresent()) {
+            long index = indexOptional.get();
             OffsetToMessageIdCacheProvider.getOrCreateCache(pulsarAdmin, brokerCluster)
                     .putMessageIdByOffset(topic, index, messageId);
             return index;
-        }).findFirst().orElseThrow(() -> new PulsarAdminException("Missing message index in " + messageId));
+        } else {
+            throw new PulsarAdminException("Message index not found for " + messageId + " in topic " + topic);
+        }
     }
 
     private static long processMessageId(String topic, MessageIdAdv messageId, String brokerCluster,

pulsar-rpc-contrib/pom.xml

@@ -31,11 +31,13 @@
         <dependency>
             <groupId>org.apache.pulsar</groupId>
             <artifactId>pulsar-client-admin</artifactId>
+            <version>${pulsar.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.pulsar</groupId>
             <artifactId>pulsar-client</artifactId>
+            <version>${pulsar.version}</version>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>