[ISSUE #137] feat: add a simple consumer mode for normal-topic consumption

.asf.yaml

@@ -10,8 +10,7 @@ github:
     issues: true
     # Enable wiki
     wiki: true
-  protected_branches:
-
+
 notifications:
   commits:      commits@rocketmq.apache.org
   issues:       commits@rocketmq.apache.org

flink-connector-rocketmq-grpc/pom.xml

@@ -26,7 +26,7 @@ under the License.
     <parent>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-connector-rocketmq-parent</artifactId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>3.0.0</version>
     </parent>
 
     <artifactId>flink-connector-rocketmq-grpc</artifactId>

flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQLiteAckClient.java → flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckClient.java

@@ -47,10 +47,12 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
- * A credential-free acknowledgement client that a downstream operator uses to acknowledge or
- * re-schedule RocketMQ Pop messages described by a {@link RocketMQReceiptHandle}. The credentials
- * used to talk to the RocketMQ proxy are configured on the operator that owns the client and are
- * <b>never</b> carried in the data stream.
+ * A credential-free acknowledgement client that acknowledges RocketMQ Pop messages of both lite and
+ * normal topics. A downstream operator uses it to acknowledge or re-schedule the messages described
+ * by a {@link RocketMQReceiptHandle}, which is what the LITE mode of the source relies on: the ack
+ * decision is only known after downstream processing, possibly on the other side of a shuffle. The
+ * credentials used to talk to the RocketMQ proxy are configured on the operator that owns the
+ * client and are <b>never</b> carried in the data stream.
  *
  * <p>The client keeps a lazily populated pool of same-group {@link LiteSimpleConsumer}s, one per
  * {@link ConsumerKey routing triple} carried by the incoming handles, and issues the {@code ack} /
@@ -59,16 +61,18 @@
  * come from the handle (so the ack RPC is routed to the right proxy and resource) and whose
  * credentials/TLS/timeout come from the operator {@link Configuration}. This is required because
  * the SDK stamps the ack request with the <em>consumer's</em> namespace, so a pooled consumer must
- * share the handle's namespace.
+ * share the handle's namespace. The pooled consumers are built through {@code LiteSimpleConsumer},
+ * but the ack RPC itself is protocol-level identical for lite and normal topic handles, so handles
+ * of normal topics are acknowledged without any semantic difference.
  *
  * <p>Instances are shared per TaskManager JVM: {@link #acquire(Configuration)} reference-counts a
  * client per distinct client configuration and {@link #release(Configuration)} closes it once no
  * operator instance references it anymore, so callers must not close a client directly.
  */
 @Internal
-public final class RocketMQLiteAckClient {
+public final class RocketMQAckClient {
 
-    private static final Logger LOG = LoggerFactory.getLogger(RocketMQLiteAckClient.class);
+    private static final Logger LOG = LoggerFactory.getLogger(RocketMQAckClient.class);
 
     /**
      * The await duration is only consulted by {@code receive}, which this ack-only client never
@@ -90,7 +94,7 @@ public final class RocketMQLiteAckClient {
 
     private volatile boolean closed = false;
 
-    private RocketMQLiteAckClient(Configuration configuration) {
+    private RocketMQAckClient(Configuration configuration) {
         this.configuration =
                 Objects.requireNonNull(configuration, "configuration should not be null");
         this.credentialsResolver = CredentialsResolvers.createFromConfiguration(configuration);
@@ -103,12 +107,12 @@ private RocketMQLiteAckClient(Configuration configuration) {
      * @param configuration the operator configuration carrying the RocketMQ client options.
      * @return the shared ack client; callers must not close it directly.
      */
-    public static synchronized RocketMQLiteAckClient acquire(Configuration configuration) {
+    public static synchronized RocketMQAckClient acquire(Configuration configuration) {
         Objects.requireNonNull(configuration, "configuration should not be null");
         final String key = keyOf(configuration);
         RefCounted ref = CLIENTS.get(key);
         if (ref == null) {
-            ref = new RefCounted(new RocketMQLiteAckClient(configuration));
+            ref = new RefCounted(new RocketMQAckClient(configuration));
             CLIENTS.put(key, ref);
         }
         ref.count++;
@@ -128,8 +132,8 @@ public static synchronized RocketMQLiteAckClient acquire(Configuration configura
     public static void release(Configuration configuration) {
         Objects.requireNonNull(configuration, "configuration should not be null");
         final String key = keyOf(configuration);
-        RocketMQLiteAckClient toClose = null;
-        synchronized (RocketMQLiteAckClient.class) {
+        RocketMQAckClient toClose = null;
+        synchronized (RocketMQAckClient.class) {
             final RefCounted ref = CLIENTS.get(key);
             if (ref == null) {
                 return;
@@ -337,10 +341,10 @@ private LiteSimpleConsumer createConsumer(ConsumerKey key, String bindTopic) {
     }
 
     private static final class RefCounted {
-        private final RocketMQLiteAckClient client;
+        private final RocketMQAckClient client;
         private int count;
 
-        private RefCounted(RocketMQLiteAckClient client) {
+        private RefCounted(RocketMQAckClient client) {
             this.client = client;
         }
     }

flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/RocketMQAckProcessFunction.java

@@ -29,13 +29,13 @@
 
 /**
  * An abstract {@link ProcessFunction} that gives subclasses a ready-to-use {@link
- * RocketMQLiteAckClient} so they can acknowledge or re-schedule RocketMQ Pop messages directly from
+ * RocketMQAckClient} so they can acknowledge or re-schedule RocketMQ Pop messages directly from
  * their own business logic via {@link #ack} and {@link #changeInvisibleDuration}.
  *
- * <p>The client is shared per TaskManager through {@link RocketMQLiteAckClient#acquire}: it is
- * acquired in {@link #open(OpenContext)} and released in {@link #close()}. The RocketMQ client
- * options (endpoints, namespace, credentials, TLS, timeout) are provided through the {@link
- * Configuration} passed to the constructor and never travel in the data stream.
+ * <p>The client is shared per TaskManager through {@link RocketMQAckClient#acquire}: it is acquired
+ * in {@link #open(OpenContext)} and released in {@link #close()}. The RocketMQ client options
+ * (endpoints, namespace, credentials, TLS, timeout) are provided through the {@link Configuration}
+ * passed to the constructor and never travel in the data stream.
  *
  * @param <IN> the input record type, typically {@code AckableMessage<T>}.
  * @param <OUT> the output record type.
@@ -47,7 +47,7 @@ public abstract class RocketMQAckProcessFunction<IN, OUT> extends ProcessFunctio
 
     private final Configuration configuration;
 
-    private transient RocketMQLiteAckClient ackClient;
+    private transient RocketMQAckClient ackClient;
     private transient boolean acquired;
     private transient Counter numAcksSucceeded;
     private transient Counter numAcksFailed;
@@ -63,7 +63,7 @@ protected RocketMQAckProcessFunction(Configuration configuration) {
     @Override
     public void open(OpenContext openContext) throws Exception {
         super.open(openContext);
-        this.ackClient = RocketMQLiteAckClient.acquire(configuration);
+        this.ackClient = RocketMQAckClient.acquire(configuration);
         this.acquired = true;
         this.numAcksSucceeded = getRuntimeContext().getMetricGroup().counter("numAcksSucceeded");
         this.numAcksFailed = getRuntimeContext().getMetricGroup().counter("numAcksFailed");
@@ -80,7 +80,7 @@ public void close() throws Exception {
         // Only release when open() actually acquired the client; Flink calls close() even after a
         // failed open(), and a stray release would decrement another operator's reference.
         if (acquired) {
-            RocketMQLiteAckClient.release(configuration);
+            RocketMQAckClient.release(configuration);
             acquired = false;
         }
         this.ackClient = null;

flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/ConsumerMode.java

@@ -0,0 +1,41 @@
+/*
+ * 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.flink.connector.rocketmq.grpc.source;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+/** The consumption mode of the RocketMQ gRPC source. */
+@PublicEvolving
+public enum ConsumerMode {
+
+    /**
+     * Bind a main lite topic via a {@code LiteSimpleConsumer}; the broker delivers messages of all
+     * sub topics over one receive stream. Messages are acknowledged by a downstream operator using
+     * the receipt handle carried by each emitted record.
+     */
+    LITE,
+
+    /**
+     * Subscribe to a normal topic via a {@code SimpleConsumer} with an optional filter expression.
+     * Messages are acknowledged by the source itself once the enclosing checkpoint completes
+     * (at-least-once); checkpointing must be enabled and the invisible duration must exceed the
+     * checkpoint interval plus timeout.
+     */
+    SIMPLE
+}

flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSource.java

@@ -39,6 +39,7 @@
 import org.apache.flink.connector.rocketmq.grpc.source.enumerator.RocketMQGrpcSourceEnumState;
 import org.apache.flink.connector.rocketmq.grpc.source.enumerator.RocketMQGrpcSourceEnumStateSerializer;
 import org.apache.flink.connector.rocketmq.grpc.source.enumerator.RocketMQGrpcSourceEnumerator;
+import org.apache.flink.connector.rocketmq.grpc.source.reader.CheckpointAckTracker;
 import org.apache.flink.connector.rocketmq.grpc.source.reader.MessageViewImpl;
 import org.apache.flink.connector.rocketmq.grpc.source.reader.RocketMQGrpcSourceFetcherManager;
 import org.apache.flink.connector.rocketmq.grpc.source.reader.RocketMQGrpcSourceReader;
@@ -116,17 +117,27 @@ public UserCodeClassLoader getUserCodeClassLoader() {
         final RocketMQGrpcSourceFetcherManager fetcherManager =
                 new RocketMQGrpcSourceFetcherManager(elementsQueue, splitReaderSupplier);
 
+        final ConsumerMode mode = configuration.get(RocketMQGrpcSourceOptions.MODE);
+        final CheckpointAckTracker ackTracker =
+                mode == ConsumerMode.SIMPLE ? new CheckpointAckTracker() : null;
+
         final RocketMQGrpcSourceRecordEmitter<OUT> recordEmitter =
                 new RocketMQGrpcSourceRecordEmitter<>(
                         deserializationSchema,
                         configuration.get(RocketMQGrpcOptions.NAMESPACE),
-                        configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP));
-
-        final RocketMQGrpcSourceReader<OUT> reader =
-                new RocketMQGrpcSourceReader<>(
-                        elementsQueue, fetcherManager, recordEmitter, configuration, readerContext);
-
-        return reader;
+                        configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP),
+                        ackTracker == null ? null : ackTracker::add);
+
+        // The same split reader instance is handed to the fetcher manager and to the source reader:
+        // in SIMPLE mode the reader acks through the very consumer that received the messages.
+        return new RocketMQGrpcSourceReader<>(
+                elementsQueue,
+                fetcherManager,
+                recordEmitter,
+                configuration,
+                readerContext,
+                ackTracker,
+                splitReader);
     }
 
     @Override

flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/source/RocketMQGrpcSourceBuilder.java

@@ -37,6 +37,8 @@ public class RocketMQGrpcSourceBuilder<OUT> {
 
     private final Configuration configuration;
     private String mainTopic;
+    private ConsumerMode mode = ConsumerMode.SIMPLE;
+    private String topic;
     private Boundedness boundedness;
     private RocketMQGrpcDeserializationSchema<OUT> deserializationSchema;
 
@@ -55,7 +57,11 @@ public RocketMQGrpcSourceBuilder<OUT> setConsumerGroup(String consumerGroup) {
         return setConfig(RocketMQGrpcSourceOptions.CONSUMER_GROUP, consumerGroup);
     }
 
-    /** Set the main lite topic bound by the LiteSimpleConsumer. Every subtask binds this topic. */
+    /**
+     * Set the main lite topic bound by the LiteSimpleConsumer. Every subtask binds this topic.
+     * Required in {@link ConsumerMode#LITE}, which must be selected explicitly via {@link
+     * #setMode(ConsumerMode)}; ignored in {@link ConsumerMode#SIMPLE}.
+     */
     public RocketMQGrpcSourceBuilder<OUT> setMainTopic(String mainTopic) {
         checkArgument(
                 mainTopic != null && !mainTopic.trim().isEmpty(),
@@ -64,6 +70,32 @@ public RocketMQGrpcSourceBuilder<OUT> setMainTopic(String mainTopic) {
         return this;
     }
 
+    /** Set the consumption mode. Defaults to {@link ConsumerMode#SIMPLE}. */
+    public RocketMQGrpcSourceBuilder<OUT> setMode(ConsumerMode mode) {
+        this.mode = checkNotNull(mode);
+        return this;
+    }
+
+    /**
+     * Set the normal topic subscribed by the SimpleConsumer. Required in the default {@link
+     * ConsumerMode#SIMPLE}; ignored in {@link ConsumerMode#LITE}.
+     */
+    public RocketMQGrpcSourceBuilder<OUT> setTopic(String topic) {
+        checkArgument(topic != null && !topic.trim().isEmpty(), "topic must not be null or blank");
+        this.topic = topic;
+        return this;
+    }
+
+    /** Set the SIMPLE mode filter expression, e.g. {@code "tagA||tagB"} or a SQL92 expression. */
+    public RocketMQGrpcSourceBuilder<OUT> setFilterExpression(String filterExpression) {
+        return setConfig(RocketMQGrpcSourceOptions.FILTER_EXPRESSION, filterExpression);
+    }
+
+    /** Set the SIMPLE mode filter expression type: {@code TAG} (default) or {@code SQL92}. */
+    public RocketMQGrpcSourceBuilder<OUT> setFilterType(String filterType) {
+        return setConfig(RocketMQGrpcSourceOptions.FILTER_TYPE, filterType);
+    }
+
     /**
      * Set the fetch concurrency of each subtask, i.e. the number of concurrent fetch requests it
      * issues. The requests are served by worker threads sharing a single thread-safe {@code
@@ -131,14 +163,23 @@ public RocketMQGrpcSource<OUT> build() {
         checkNotNull(
                 configuration.get(RocketMQGrpcSourceOptions.CONSUMER_GROUP),
                 "consumer group must be configured");
-        checkArgument(
-                mainTopic != null && !mainTopic.trim().isEmpty(),
-                "the main topic must be configured");
+        checkNotNull(mode, "the consumer mode must be configured");
+        if (mode == ConsumerMode.LITE) {
+            checkArgument(
+                    mainTopic != null && !mainTopic.trim().isEmpty(),
+                    "the main topic must be configured");
+            configuration.set(RocketMQGrpcSourceOptions.MAIN_TOPIC, mainTopic);
+        } else {
+            checkArgument(
+                    topic != null && !topic.trim().isEmpty(),
+                    "the topic must be configured in SIMPLE mode");
+            configuration.set(RocketMQGrpcSourceOptions.TOPIC, topic);
+        }
         checkArgument(
                 configuration.get(RocketMQGrpcSourceOptions.FETCH_CONCURRENCY) >= 1,
                 "fetch concurrency must be at least 1");
         checkNotNull(deserializationSchema, "deserializer must be configured");
-        configuration.set(RocketMQGrpcSourceOptions.MAIN_TOPIC, mainTopic);
+        configuration.set(RocketMQGrpcSourceOptions.MODE, mode);
         return new RocketMQGrpcSource<>(configuration, boundedness, deserializationSchema);
     }
 }