Initial implementation

Author: aaron-ai

.github/workflows/java_coverage.yml

@@ -19,7 +19,8 @@ jobs:
           java-version: "11"
           distribution: "adopt"
       - name: Generate coverage report
-        run: mvn test --file ./java/pom.xml
+        working-directory: ./java
+        run: mvn test --file ./pom.xml
       - name: Test summary
         uses: test-summary/action@v1
         with:

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/ClientServiceProvider.java

@@ -19,6 +19,7 @@
 
 import java.util.Iterator;
 import java.util.ServiceLoader;
+import org.apache.rocketmq.client.apis.consumer.PullConsumerBuilder;
 import org.apache.rocketmq.client.apis.consumer.PushConsumerBuilder;
 import org.apache.rocketmq.client.apis.consumer.SimpleConsumerBuilder;
 import org.apache.rocketmq.client.apis.message.MessageBuilder;
@@ -65,4 +66,11 @@ static ClientServiceProvider loadService() {
      * @return the simple consumer builder instance.
      */
     SimpleConsumerBuilder newSimpleConsumerBuilder();
+
+    /**
+     * Get the pull consumer builder by the current provider.
+     *
+     * @return the pull consumer builder instance.
+     */
+    PullConsumerBuilder newPullConsumerBuilder();
 }

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/PullConsumer.java

@@ -0,0 +1,134 @@
+/*
+ * 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.rocketmq.client.apis.consumer;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.apis.message.MessageQueue;
+import org.apache.rocketmq.client.apis.message.MessageView;
+
+public interface PullConsumer extends Closeable {
+    /**
+     * Get the consumer group of the consumer.
+     */
+    String getConsumerGroup();
+
+    /**
+     * @param topic    the topic that needs to be monitored.
+     * @param listener the callback to detect the message queue changes.
+     */
+    void registerMessageQueueChangeListenerByTopic(String topic, TopicMessageQueueChangeListener listener);
+
+    /**
+     * Fetch message queues of the topic.
+     */
+    Collection<MessageQueue> fetchMessageQueues(String topic) throws ClientException;
+
+    /**
+     * Manually assign a list of message queues to this consumer.
+     *
+     * <p>This interface does not allow for incremental assignment and will replace the previous assignment (if
+     * previous assignment existed).
+     *
+     * @param messageQueues the list of message queues that are to be assigned to this consumer.
+     */
+    void assign(Collection<MessageQueue> messageQueues);
+
+    /**
+     * Fetch messages from assigned message queues specified by {@link #assign(Collection)}.
+     *
+     * @param timeout the maximum time to block.
+     * @return list of fetched messages.
+     */
+    List<MessageView> poll(Duration timeout) throws InterruptedException;
+
+    /**
+     * Overrides the fetch offsets that the consumer will use on the next poll. If this method is invoked for the same
+     * message queue more than once, the latest offset will be used on the next {@link #poll(Duration)}.
+     *
+     * @param messageQueue the message queue to override the fetch offset.
+     * @param offset       message offset.
+     */
+    void seek(MessageQueue messageQueue, long offset);
+
+    /**
+     * Suspending message pulling from the message queues.
+     *
+     * @param messageQueues message queues that need to be suspended.
+     */
+    void pause(Collection<MessageQueue> messageQueues);
+
+    /**
+     * Resuming message pulling from the message queues.
+     *
+     * @param messageQueues message queues that need to be resumed.
+     */
+    void resume(Collection<MessageQueue> messageQueues);
+
+    /**
+     * Look up the offsets for the given message queue by timestamp. The returned offset for each message queue is the
+     * earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding message
+     * queue.
+     *
+     * @param messageQueue message queue that needs to be looked up.
+     * @param timestamp    the timestamp for which to search.
+     * @return the offset of the message queue, or {@link Optional#empty()} if there is no message.
+     */
+    Optional<Long> offsetForTimestamp(MessageQueue messageQueue, long timestamp) throws ClientException;
+
+    /**
+     * Get the latest committed offset for the given message queue.
+     *
+     * @return the latest committed offset, or {@link Optional#empty()} if there was no prior commit.
+     */
+    Optional<Long> committed(MessageQueue messageQueue) throws ClientException;
+
+    /**
+     * Commit offset manually.
+     */
+    void commit() throws ClientException;
+
+    /**
+     * Overrides the fetch offsets with the beginning offset that the consumer will use on the next poll. If this
+     * method is invoked for the same message queue more than once, the latest offset will be used on the next
+     * {@link #poll(Duration)}.
+     *
+     * @param messageQueue the message queue to seek.
+     */
+    void seekToBegin(MessageQueue messageQueue) throws ClientException;
+
+    /**
+     * Overrides the fetch offsets with the end offset that the consumer will use on the next poll. If this method is
+     * invoked for the same message queue more than once, the latest offset will be used on the next
+     * {@link #poll(Duration)}.
+     *
+     * @param messageQueue the message queue to seek.
+     */
+    void seekToEnd(MessageQueue messageQueue) throws ClientException;
+
+    /**
+     * Close the pull consumer and release all related resources.
+     */
+    @Override
+    void close() throws IOException;
+}

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/PullConsumerBuilder.java

@@ -0,0 +1,75 @@
+/*
+ * 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.rocketmq.client.apis.consumer;
+
+import java.time.Duration;
+import org.apache.rocketmq.client.apis.ClientConfiguration;
+import org.apache.rocketmq.client.apis.ClientException;
+
+public interface PullConsumerBuilder {
+    /**
+     * Set the client configuration for the consumer.
+     *
+     * @param clientConfiguration client's configuration.
+     * @return the consumer builder instance.
+     */
+    PullConsumerBuilder setClientConfiguration(ClientConfiguration clientConfiguration);
+
+    /**
+     * Set the load balancing group for the consumer.
+     *
+     * @param consumerGroup consumer load balancing group.
+     * @return the consumer builder instance.
+     */
+    PullConsumerBuilder setConsumerGroup(String consumerGroup);
+
+    /**
+     * Set the consumer's offset commit interval if auto commit is enabled.
+     *
+     * @param duration offset commit interval
+     * @return the consumer builder instance.
+     */
+    PullConsumerBuilder setAutoCommitInterval(Duration duration);
+
+    /**
+     * Set the maximum number of messages cached locally.
+     *
+     * @param count message count.
+     * @return the consumer builder instance.
+     */
+    PullConsumerBuilder setMaxCacheMessageCountEachQueue(int count);
+
+    /**
+     * Set the maximum bytes of messages cached locally.
+     *
+     * @param bytes message size.
+     * @return the consumer builder instance.
+     */
+    PullConsumerBuilder setMaxCacheMessageSizeInBytesEachQueue(int bytes);
+
+    /**
+     * Finalize the build of {@link PullConsumer} and start.
+     *
+     * <p>This method will block until the pull consumer starts successfully.
+     *
+     * <p>Especially, if this method is invoked more than once, different pull consumer will be created and started.
+     *
+     * @return the pull consumer instance.
+     */
+    PullConsumer build() throws ClientException;
+}

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/consumer/TopicMessageQueueChangeListener.java

@@ -0,0 +1,32 @@
+/*
+ * 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.rocketmq.client.apis.consumer;
+
+import java.util.Set;
+import org.apache.rocketmq.client.apis.message.MessageQueue;
+
+public interface TopicMessageQueueChangeListener {
+    /**
+     * This method will be invoked in the condition of queue numbers changed, These scenarios occur when the topic is
+     * expanded or shrunk.
+     *
+     * @param topic         the topic to listen.
+     * @param messageQueues latest message queues of the topic.
+     */
+    void onChanged(String topic, Set<MessageQueue> messageQueues);
+}

java/client-apis/src/main/java/org/apache/rocketmq/client/apis/message/MessageQueue.java

@@ -0,0 +1,30 @@
+/*
+ * 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.rocketmq.client.apis.message;
+
+public interface MessageQueue {
+    /**
+     * Topic of the current message queue.
+     */
+    String getTopic();
+
+    /**
+     * Get the identifier of the current message queue.
+     */
+    String getId();
+}

java/client/src/main/java/org/apache/rocketmq/client/java/example/PullConsumerExample.java

@@ -0,0 +1,74 @@
+/*
+ * 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.rocketmq.client.java.example;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import org.apache.rocketmq.client.apis.ClientConfiguration;
+import org.apache.rocketmq.client.apis.ClientException;
+import org.apache.rocketmq.client.apis.ClientServiceProvider;
+import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
+import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
+import org.apache.rocketmq.client.apis.consumer.PullConsumer;
+import org.apache.rocketmq.client.apis.message.MessageQueue;
+import org.apache.rocketmq.client.apis.message.MessageView;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PullConsumerExample {
+    private static final Logger log = LoggerFactory.getLogger(PullConsumerExample.class);
+
+    private PullConsumerExample() {
+    }
+
+    @SuppressWarnings({"resource", "InfiniteLoopStatement"})
+    public static void main(String[] args) throws ClientException, InterruptedException {
+        final ClientServiceProvider provider = ClientServiceProvider.loadService();
+
+        // Credential provider is optional for client configuration.
+        String accessKey = "yourAccessKey";
+        String secretKey = "yourSecretKey";
+        SessionCredentialsProvider sessionCredentialsProvider =
+            new StaticSessionCredentialsProvider(accessKey, secretKey);
+
+        String endpoints = "foobar.com:8080";
+        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
+            .setEndpoints(endpoints)
+            .setCredentialProvider(sessionCredentialsProvider)
+            .build();
+        String topic = "yourTopic";
+        final PullConsumer consumer = provider.newPullConsumerBuilder()
+            .setClientConfiguration(clientConfiguration)
+            .setAutoCommitInterval(Duration.ofSeconds(3))
+            .build();
+
+        final Collection<MessageQueue> mqs = consumer.fetchMessageQueues(topic);
+        consumer.assign(mqs);
+
+        do {
+            final List<MessageView> messages = consumer.poll(Duration.ofSeconds(5));
+            log.info("Pulled {} message(s)", messages.size());
+            for (MessageView message : messages) {
+                log.info("message is pulled, messageId={}", message.getMessageId());
+            }
+        } while (true);
+        // Close the pull consumer when you don't need it anymore.
+        // consumer.close();
+    }
+}