apache · lizhimins · Jul 22, 2026
diff --git a/README.md b/README.md
@@ -11,8 +11,10 @@ Two connector tracks are shipped:
 | Track | Module | RocketMQ client | Best for |
 | --- | --- | --- | --- |
 | **Remoting** | `flink-connector-rocketmq` | `rocketmq-client` (remoting protocol) | Classic topics, FLIP-27 source + SinkV2 sink, SQL |
+| **gRPC** | `flink-connector-rocketmq-grpc` | `rocketmq-client-java` (gRPC protocol) | RocketMQ 5.x lite topics, downstream ack / fair throttling |
 
-The SQL fat-jar is packaged by `flink-sql-connector-rocketmq`.
+The SQL fat-jars are packaged by `flink-sql-connector-rocketmq` and
+`flink-sql-connector-rocketmq-grpc`.
 
 ## Apache Flink
 
@@ -93,13 +95,28 @@ CREATE TABLE rocketmq_source (
 );
 ```
 
+### gRPC lite consumer (RocketMQ 5.x)
+
+```java
+RocketMQGrpcSource<String> source = RocketMQGrpcSource.<String>builder()
+        .setEndpoints("127.0.0.1:8081")
+        .setConsumerGroup("GID-lite")
+        .setMainTopic("LiteMainTopic")        // wildcard-subscribes all lite topics under it
+        .setValueOnlyDeserializer(new SimpleStringSchema())
+        .build();
+```
+
+The gRPC source emits `AckableMessage<OUT>` and never acks by itself: the downstream operators own
+the ack / throttle decision. See the gRPC connector doc for the ack operator wirings.
+
 ## Documentation
 
 Connector documentation is located in the `docs/` directory of this repository:
 
 | Document | Content |
 | --- | --- |
 | [docs/remoting-connector.md](docs/remoting-connector.md) | Remoting DataStream connector: usage and all configuration options |
+| [docs/grpc-connector.md](docs/grpc-connector.md) | gRPC lite connector: prerequisites, downstream ack / throttling, options |
 | [docs/sql-connector.md](docs/sql-connector.md) | Table/SQL connector: DDL, metadata columns, fat-jar notes |
 | [docs/legacy-connector.md](docs/legacy-connector.md) | Legacy `RocketMQSourceFunction` / `RocketMQSink` (deprecated) |
 | [docs/connector-overview.md](docs/connector-overview.md) | Feature comparison with other messaging connectors |

diff --git a/docs/grpc-connector.md b/docs/grpc-connector.md
diff --git a/docs/sql-connector.md b/docs/sql-connector.md
@@ -1,7 +1,9 @@
 # Table / SQL Connector
 
 The remoting SQL connector is packaged as the fat-jar `flink-sql-connector-rocketmq`
-(identifier `rocketmq`).
+(identifier `rocketmq`). The gRPC SQL connector is `flink-sql-connector-rocketmq-grpc`
+(identifier `rocketmq-grpc`); both fat-jars relocate their dependencies and can be deployed on
+the same classpath.
 
 ## Creating tables
 

diff --git a/flink-connector-rocketmq-grpc/pom.xml b/flink-connector-rocketmq-grpc/pom.xml
@@ -0,0 +1,153 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.flink</groupId>
+        <artifactId>flink-connector-rocketmq-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>flink-connector-rocketmq-grpc</artifactId>
+    <name>Flink : Connectors : RocketMQ gRPC</name>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <!-- Core -->
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- Table ecosystem (optional, for Table API integration) -->
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-common</artifactId>
+            <scope>provided</scope>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-api-java-bridge</artifactId>
+            <scope>provided</scope>
+            <optional>true</optional>
+        </dependency>
+
+        <!-- RocketMQ gRPC client (5.x) -->
+
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client-java</artifactId>
+        </dependency>
+
+        <!-- Tests -->
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-test-utils</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-test-utils</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-common</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-runtime</artifactId>
+            <scope>test</scope>
+            <type>test-jar</type>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <id>analyze-deps</id>
+                        <goals>
+                            <goal>analyze</goal>
+                        </goals>
+                        <phase>verify</phase>
+                        <configuration>
+                            <ignoredUnusedDeclaredDependencies>
+                                org.apache.flink:flink-table-api-java-bridge
+                            </ignoredUnusedDeclaredDependencies>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>
diff --git a/...etmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcOptions.java b/...etmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcOptions.java
@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import java.time.Duration;
+
+/**
+ * Shared configuration options for the RocketMQ gRPC connector. These options describe how the
+ * underlying {@code rocketmq-client-java} SDK connects to the RocketMQ proxy (endpoints,
+ * credentials, TLS and namespace) and are consumed by both the source and the sink.
+ *
+ * <p>These are the programmatic/SDK-facing options used by the source and sink builders and their
+ * runtime; every key carries the {@link #CLIENT_CONFIG_PREFIX} prefix. The SQL DDL keys are defined
+ * separately in {@code RocketMQGrpcConnectorOptions}.
+ */
+@PublicEvolving
+public class RocketMQGrpcOptions {
+
+    private RocketMQGrpcOptions() {}
+
+    /** Prefix for the shared RocketMQ gRPC client options. */
+    public static final String CLIENT_CONFIG_PREFIX = "rocketmq.client.";
+
+    public static final ConfigOption<String> ENDPOINTS =
+            ConfigOptions.key(CLIENT_CONFIG_PREFIX + "endpoints")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The access point (proxy) endpoints the gRPC SDK communicates with, "
+                                    + "for example '127.0.0.1:8080'.");
+
+    public static final ConfigOption<String> NAMESPACE =
+            ConfigOptions.key(CLIENT_CONFIG_PREFIX + "namespace")
+                    .stringType()
+                    .defaultValue("")
+                    .withDescription("The resource namespace of the RocketMQ instance.");
+
+    public static final ConfigOption<String> ACCESS_KEY =
+            ConfigOptions.key(CLIENT_CONFIG_PREFIX + "access-key")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The access key used for static session credentials.");
+
+    public static final ConfigOption<String> SECRET_KEY =
+            ConfigOptions.key(CLIENT_CONFIG_PREFIX + "secret-key")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription("The secret key used for static session credentials.");
+
+    /**
+     * The fully qualified class name of a {@link
+     * org.apache.flink.connector.rocketmq.grpc.common.CredentialsResolver} implementation (public,
+     * with a public no-argument constructor).
+     *
+     * <p>The resolver is instantiated reflectively on each TaskManager (never serialized) and
+     * resolves session credentials per proxy endpoint locally, e.g. from environment variables,
+     * mounted secret files or an external KMS. This keeps plaintext secrets out of the job graph
+     * and lets one downstream ack operator serve handles from multiple clusters with different
+     * credentials. When set, it takes precedence over {@link #ACCESS_KEY} / {@link #SECRET_KEY};
+     * returning {@code null} for an endpoint means no authentication is required.
+     */
+    public static final ConfigOption<String> CREDENTIALS_RESOLVER_CLASS =
+            ConfigOptions.key(CLIENT_CONFIG_PREFIX + "credentials-resolver-class")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "The fully qualified class name of a CredentialsResolver that resolves "
+                                    + "session credentials per endpoint locally on the TaskManager "
+                                    + "(for example from environment variables, mounted secret files "
+                                    + "or an external KMS). When set, it takes precedence over the "
+                                    + "static access-key/secret-key options and keeps plaintext "
+                                    + "credentials out of the job graph.");
+
+    public static final ConfigOption<Boolean> TLS_ENABLED =
+            ConfigOptions.key(CLIENT_CONFIG_PREFIX + "tls-enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Whether TLS is enabled for the gRPC transport.");
+
+    public static final ConfigOption<Duration> REQUEST_TIMEOUT =
+            ConfigOptions.key(CLIENT_CONFIG_PREFIX + "request-timeout")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(3))
+                    .withDescription("The request timeout for a single gRPC invocation.");
+}
diff --git a/...ketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/AckableMessage.java b/...ketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/AckableMessage.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.flink.connector.rocketmq.grpc.ack;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.Objects;
+
+/**
+ * The element type produced by the RocketMQ gRPC source when running in downstream-acknowledgement
+ * mode. It pairs the deserialized record {@code value} with the credential-free {@link
+ * RocketMQReceiptHandle} needed to acknowledge (or re-schedule) the originating message from any
+ * downstream operator.
+ *
+ * @param <T> the deserialized record type.
+ */
+@PublicEvolving
+public final class AckableMessage<T> {
+
+    private final T value;
+    private final RocketMQReceiptHandle handle;
+
+    public AckableMessage(T value, RocketMQReceiptHandle handle) {
+        this.value = value;
+        this.handle = Objects.requireNonNull(handle, "handle should not be null");
+    }
+
+    /** The deserialized record value. */
+    public T getValue() {
+        return value;
+    }
+
+    /** The receipt handle used to acknowledge or re-schedule the originating message. */
+    public RocketMQReceiptHandle getHandle() {
+        return handle;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        AckableMessage<?> that = (AckableMessage<?>) o;
+        return Objects.equals(value, that.value) && handle.equals(that.handle);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(value, handle);
+    }
+
+    @Override
+    public String toString() {
+        return "AckableMessage{value=" + value + ", handle=" + handle + '}';
+    }
+}