This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-flink.git
The following commit(s) were added to refs/heads/main by this push:
new 3770862 [ISSUE #140] build: upgrade the RocketMQ client to 5.5.1 and
fix the fat jar shading (#144)
3770862 is described below
commit 377086260c0bc98f4a946c6f72b29d0fd886ac8f
Author: lizhimins <[email protected]>
AuthorDate: Mon Oct 26 14:37:30 2026 +0800
[ISSUE #140] build: upgrade the RocketMQ client to 5.5.1 and fix the fat
jar shading (#144)
Move rocketmq-client from 5.1.1 to the latest community release 5.5.1 and
rocketmq-client-java from 5.2.1 to 5.2.2.
rocketmq-acl was retired in 5.5.x: AclClientRPCHook and SessionCredentials
now
ship inside rocketmq-client under the same package, so the extra dependency
is
dropped. RocketMQ also replaced fastjson with fastjson2, which broke the
three
logging call sites; they now use fastjson2, and the dependency is declared
explicitly instead of being inherited transitively. rocketmq-filter is
added for
the SQL92 evaluator the source needs.
Shading: the artifact set did not match the relocations, so references were
rewritten to shaded names whose classes were never bundled. Bundling is now
derived from what the relocations rewrite. Both fat jars also stop
relocating
the logging packages, because rmq.client.logback.xml and
rocketmq.logback.xml
reference their appenders by name in plain text and shade does not rewrite
XML
resources: relocating them left every appender pointing at a class that no
longer exists, so the remoting jar died at startup and the gRPC jar silently
lost all client log output.
---
flink-connector-rocketmq/pom.xml | 7 ++-
.../rocketmq/source/InnerConsumerImpl.java | 2 +-
.../RocketMQSourceEnumStateSerializer.java | 2 +-
.../enumerator/RocketMQSourceEnumerator.java | 4 +-
flink-sql-connector-rocketmq-grpc/pom.xml | 15 ++++++
flink-sql-connector-rocketmq/pom.xml | 53 +++++++++++++++++++++-
pom.xml | 29 ++++++++++--
7 files changed, 102 insertions(+), 10 deletions(-)
diff --git a/flink-connector-rocketmq/pom.xml b/flink-connector-rocketmq/pom.xml
index b022b86..2a2565b 100644
--- a/flink-connector-rocketmq/pom.xml
+++ b/flink-connector-rocketmq/pom.xml
@@ -86,7 +86,12 @@ under the License.
<dependency>
<groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-acl</artifactId>
+ <artifactId>rocketmq-filter</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.alibaba.fastjson2</groupId>
+ <artifactId>fastjson2</artifactId>
</dependency>
<dependency>
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
index 8c79e0f..1149de6 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/InnerConsumerImpl.java
@@ -27,7 +27,7 @@ import
org.apache.flink.connector.rocketmq.source.util.UtilAll;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.StringUtils;
-import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson2.JSON;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumStateSerializer.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumStateSerializer.java
index 960e622..be3f5f6 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumStateSerializer.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumStateSerializer.java
@@ -20,7 +20,7 @@ package org.apache.flink.connector.rocketmq.source.enumerator;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson2.JSON;
import org.apache.rocketmq.common.message.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
index 45b1132..12e31f7 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/source/enumerator/RocketMQSourceEnumerator.java
@@ -42,7 +42,7 @@ import
org.apache.flink.connector.rocketmq.source.enumerator.offset.OffsetsSelec
import org.apache.flink.connector.rocketmq.source.split.RocketMQSourceSplit;
import org.apache.flink.util.FlinkRuntimeException;
-import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson2.JSON;
import org.apache.rocketmq.common.message.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -572,7 +572,7 @@ public class RocketMQSourceEnumerator
if (!incrementalSplit.isEmpty()) {
log.info(
"Enumerator assigning split(s) to readers {}",
- JSON.toJSONString(incrementalSplit, false));
+ JSON.toJSONString(incrementalSplit));
context.assignSplits(new SplitsAssignment<>(incrementalSplit));
}
diff --git a/flink-sql-connector-rocketmq-grpc/pom.xml
b/flink-sql-connector-rocketmq-grpc/pom.xml
index 92444ab..cddedc4 100644
--- a/flink-sql-connector-rocketmq-grpc/pom.xml
+++ b/flink-sql-connector-rocketmq-grpc/pom.xml
@@ -110,6 +110,21 @@ under the License.
<relocation>
<pattern>org.apache.rocketmq</pattern>
<shadedPattern>org.apache.flink.rocketmq.grpc.shaded.org.apache.rocketmq</shadedPattern>
+ <!-- rocketmq-client-java is already
shaded upstream: its
+ logback lives under
org.apache.rocketmq.shaded.ch.qos.logback
+ and its own appenders under
+
org.apache.rocketmq.client.java.logging. The bundled
+ rocketmq.logback.xml references both
by name in plain text
+ and shade never rewrites class names
inside resources, so
+ relocating them makes every appender
fail to load and the
+ gRPC client produces no log output at
all. Keep those names,
+ exactly like the remoting fat-jar
keeps
+ org.apache.rocketmq.logging. -->
+ <excludes>
+
<exclude>org.apache.rocketmq.logging.**</exclude>
+
<exclude>org.apache.rocketmq.shaded.ch.qos.logback.**</exclude>
+
<exclude>org.apache.rocketmq.client.java.logging.**</exclude>
+ </excludes>
</relocation>
</relocations>
<transformers>
diff --git a/flink-sql-connector-rocketmq/pom.xml
b/flink-sql-connector-rocketmq/pom.xml
index ad3bc8f..f1b32b4 100644
--- a/flink-sql-connector-rocketmq/pom.xml
+++ b/flink-sql-connector-rocketmq/pom.xml
@@ -87,10 +87,48 @@ under the License.
<includes>
<include>org.apache.flink:flink-connector-rocketmq</include>
<include>org.apache.rocketmq:*</include>
-
<include>org.apache.rocketmq:schema-registry-client</include>
-
<include>commons-lang:commons-lang</include>
+ <!-- RocketMQ 5.x ships its logging
implementation under the
+ io.github.aliyunmq group (package is
still
+ org.apache.rocketmq.logging.*), so
the org.apache.rocketmq:*
+ entry above does not cover it.
Without it the client fails at
+ logger initialization with
ClassNotFoundException. -->
+ <include>io.github.aliyunmq:*</include>
+ <!-- The groups below are all rewritten by
<relocations>, so they
+ must be bundled as well: a relocation
only rewrites references,
+ it does not move classes into the
jar. Rewriting a reference to
+ a shaded name without shipping the
class is a guaranteed
+ ClassNotFoundException at runtime. -->
+ <include>io.netty:*</include>
+ <include>io.grpc:*</include>
+ <include>io.perfmark:perfmark-api</include>
+ <include>com.google.protobuf:*</include>
+
<include>com.google.code.gson:gson</include>
<include>com.google.guava:guava</include>
<include>com.google.guava:failureaccess</include>
+ <!-- Imported directly by the connector's
own code. These are not
+ relocated, so they are bundled under
their original names. -->
+
<include>com.alibaba.fastjson2:fastjson2</include>
+
<include>org.apache.commons:commons-lang3</include>
+
<include>commons-lang:commons-lang</include>
+
<include>commons-collections:commons-collections</include>
+ <!-- Runtime dependencies of
rocketmq-client / rocketmq-tools -->
+
<include>commons-codec:commons-codec</include>
+
<include>commons-validator:commons-validator</include>
+
<include>commons-beanutils:commons-beanutils</include>
+
<include>commons-digester:commons-digester</include>
+
<include>commons-logging:commons-logging</include>
+ <include>commons-cli:commons-cli</include>
+ <include>commons-io:commons-io</include>
+
<include>com.googlecode.concurrentlinkedhashmap:concurrentlinkedhashmap-lru</include>
+ <!-- Deliberately not bundled: the heavy
transitive tail of
+ schema-registry-client (avro /
jackson / okhttp /
+ opentelemetry / kotlin / dozer /
snakeyaml / zstd / lz4 /
+ commons-compress), none of which the
connector imports
+ directly; plus test libraries that
leak in as compile-scope
+ transitive deps (hamcrest /
awaitility / aspectjweaver) and
+ annotation-only jars. Bundling them
would only grow the jar
+ and widen the conflict surface. Note
that RocketMQCatalog may
+ still need some of them when it talks
to a schema registry. -->
</includes>
</artifactSet>
<filters>
@@ -114,6 +152,17 @@ under the License.
<relocation>
<pattern>org.apache.rocketmq</pattern>
<shadedPattern>org.apache.flink.rocketmq.shaded.org.apache.rocketmq</shadedPattern>
+ <!-- rmq.client.logback.xml in
rocketmq-client references the
+
org.apache.rocketmq.logging.ch.qos.logback.* appenders by
+ name in plain text. Shade rewrites
class files but not class
+ names inside XML resources, so
relocating the logging
+ packages leaves those strings
pointing at classes that no
+ longer exist: the job dies at startup
with
+ ClassNotFoundException /
IncompatibleClassException.
+ The logging packages must keep their
original names. -->
+ <excludes>
+
<exclude>org.apache.rocketmq.logging.**</exclude>
+ </excludes>
</relocation>
<relocation>
<pattern>io.grpc</pattern>
diff --git a/pom.xml b/pom.xml
index 377451f..3bd4118 100644
--- a/pom.xml
+++ b/pom.xml
@@ -83,9 +83,10 @@ under the License.
<!-- Main Dependencies -->
<flink.version>1.20.3</flink.version>
- <rocketmq.version>5.1.1</rocketmq.version>
- <rocketmq.client.java.version>5.2.1</rocketmq.client.java.version>
+ <rocketmq.version>5.5.1</rocketmq.version>
+ <rocketmq.client.java.version>5.2.2</rocketmq.client.java.version>
<rocketmq.schema.registry.version>0.1.0</rocketmq.schema.registry.version>
+ <fastjson2.version>2.0.63</fastjson2.version>
<commons-lang.version>2.6</commons-lang.version>
<!-- Other Dependencies -->
@@ -187,11 +188,25 @@ under the License.
</exclusions>
</dependency>
+ <!-- rocketmq-acl was retired in 5.5.x: AclClientRPCHook and
SessionCredentials now
+ ship inside rocketmq-client under the same
org.apache.rocketmq.acl.common package. -->
+
+ <!-- SQL92 expression evaluation. The lite-pull consumer can only
carry a tag
+ sub-expression in assign mode, so a SQL92 filter has to be
evaluated against the
+ message properties on the client side. -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
- <artifactId>rocketmq-acl</artifactId>
+ <artifactId>rocketmq-filter</artifactId>
<version>${rocketmq.version}</version>
<exclusions>
+ <exclusion>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-srvutil</artifactId>
+ </exclusion>
<exclusion>
<groupId>io.github.aliyunmq</groupId>
<artifactId>rocketmq-logback-classic</artifactId>
@@ -250,6 +265,14 @@ under the License.
<version>${rocketmq.client.java.version}</version>
</dependency>
+ <!-- RocketMQ 5.5.x moved from fastjson to fastjson2. Declared
explicitly because the
+ connector imports it directly rather than relying on the
transitive path. -->
+ <dependency>
+ <groupId>com.alibaba.fastjson2</groupId>
+ <artifactId>fastjson2</artifactId>
+ <version>${fastjson2.version}</version>
+ </dependency>
+
<!-- Flink dependencies -->
<dependency>
<groupId>org.apache.flink</groupId>
