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 7fe6ad8 [ISSUE #155] refactor: target Flink 2.2.1 on main (Java 17)
(#158)
7fe6ad8 is described below
commit 7fe6ad89a17a46b640424c2f5ea219d804ceaae4
Author: lizhimins <[email protected]>
AuthorDate: Tue Nov 3 10:01:12 2026 +0800
[ISSUE #155] refactor: target Flink 2.2.1 on main (Java 17) (#158)
Now that the codebase is new-API only, supporting Flink 2.x is just the
version switch:
- flink.version 1.20.3 -> 2.2.1, Java 17 bytecode.
- The CI build matrix drops JDK 11 (Flink 2.x requires Java 17 at compile
time), but spotless 2.27.1 silently skips its Java formatter on JDK 16+
("Spotless check skipped"), so the style check moves to a dedicated
JDK 11 job. Leaving it inside the build job would have dropped the only
style gate that actually runs.
- Drop the deprecated-overload bridges that only 1.20 needs
(Sink.InitContext, createSerializer(ExecutionConfig)); 2.x declares
the WriterInitContext / SerializerConfig overloads abstract.
- The grpc ITCase collects with sink.legacy.RichSinkFunction, the 2.x
package; 1.20 has no sink.legacy subpackage.
The 1.x maintenance branch (v3.0) keeps 1.20.3 and the bridges; the diff
between the two lines is exactly this commit, so fixes cherry-pick both
ways without adaptation in the shared code.
Verified: mvn clean package green on JDK 17 (remoting 146 / grpc 80 unit
tests), spotless:check green on JDK 11, both SQL fat-jars produced.
---
.github/workflows/build.yml | 27 ++++++--
.../rocketmq/grpc/ack/AckableMessageTypeInfo.java | 8 ---
.../rocketmq/grpc/sink/RocketMQGrpcSink.java | 14 ----
.../rocketmq/grpc/RocketMQGrpcITCase.java | 2 +-
.../connector/rocketmq/sink/RocketMQSink.java | 81 ----------------------
pom.xml | 19 ++---
6 files changed, 35 insertions(+), 116 deletions(-)
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 7f8acaa..6a32a4d 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -29,6 +29,27 @@ concurrency:
cancel-in-progress: true
jobs:
+ style:
+ # spotless-maven-plugin 2.27.1 silently skips its Java formatter on JDK 16+
+ # (the log says "Spotless check skipped"), so a spotless:check step inside
the
+ # JDK 17 build job enforces nothing. Dropping JDK 11 from the build matrix
+ # would therefore drop the only real style gate, so keep it as its own job.
+ # Formatting does not compile against Flink, so it can stay on JDK 11.
+ name: Check code style (JDK 11)
+ runs-on: ubuntu-latest
+ timeout-minutes: 20
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up JDK 11
+ uses: actions/setup-java@v4
+ with:
+ distribution: temurin
+ java-version: 11
+
+ - name: Check code style
+ run: mvn -B -ntp spotless:check
+
build:
name: Build (JDK ${{ matrix.java }})
runs-on: ubuntu-latest
@@ -36,7 +57,8 @@ jobs:
strategy:
fail-fast: false
matrix:
- java: [11, 17]
+ # Flink 2.x requires Java 17 at compile time.
+ java: [17]
steps:
- uses: actions/checkout@v4
@@ -47,9 +69,6 @@ jobs:
java-version: ${{ matrix.java }}
cache: maven
- - name: Check code style
- run: mvn -B -ntp spotless:check
-
- name: Build and run tests
# `package` runs unit tests and the shade plugin for the SQL fat-jars.
# dependency:analyze (bound to `verify`) is skipped until declarations
are cleaned up.
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/AckableMessageTypeInfo.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/AckableMessageTypeInfo.java
index 336baec..f876ecf 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/AckableMessageTypeInfo.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/ack/AckableMessageTypeInfo.java
@@ -20,7 +20,6 @@ package org.apache.flink.connector.rocketmq.grpc.ack;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.serialization.SerializerConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
@@ -86,13 +85,6 @@ public final class AckableMessageTypeInfo<T> extends
TypeInformation<AckableMess
return new Serializer<>(valueTypeInfo.createSerializer(config));
}
- /** Flink 1.x still declares the deprecated {@link ExecutionConfig}
overload abstract. */
- @Override
- @Deprecated
- public TypeSerializer<AckableMessage<T>> createSerializer(ExecutionConfig
config) {
- return new Serializer<>(valueTypeInfo.createSerializer(config));
- }
-
@Override
public String toString() {
return "AckableMessageTypeInfo<" + valueTypeInfo + '>';
diff --git
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/sink/RocketMQGrpcSink.java
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/sink/RocketMQGrpcSink.java
index 50e1d1d..9c2b131 100644
---
a/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/sink/RocketMQGrpcSink.java
+++
b/flink-connector-rocketmq-grpc/src/main/java/org/apache/flink/connector/rocketmq/grpc/sink/RocketMQGrpcSink.java
@@ -58,18 +58,4 @@ public class RocketMQGrpcSink<IN> implements Sink<IN> {
serializationSchema,
context.asSerializationSchemaInitializationContext());
}
-
- /**
- * Flink 1.x still declares the deprecated {@link Sink.InitContext}
overload abstract. The 1.x
- * runtime calls {@link #createWriter(WriterInitContext)}; this bridge
serves callers that pass
- * the old context type.
- */
- @Override
- @Deprecated
- public SinkWriter<IN> createWriter(Sink.InitContext context) throws
IOException {
- return new RocketMQGrpcSinkWriter<>(
- configuration,
- serializationSchema,
- context.asSerializationSchemaInitializationContext());
- }
}
diff --git
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
index c041f9f..80ceddf 100644
---
a/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
+++
b/flink-connector-rocketmq-grpc/src/test/java/org/apache/flink/connector/rocketmq/grpc/RocketMQGrpcITCase.java
@@ -467,7 +467,7 @@ class RocketMQGrpcITCase extends AbstractTestBase {
*/
@SuppressWarnings("deprecation")
private static class CollectingSinkFunction
- extends
org.apache.flink.streaming.api.functions.sink.RichSinkFunction<String> {
+ extends
org.apache.flink.streaming.api.functions.sink.legacy.RichSinkFunction<String> {
private static final long serialVersionUID = 1L;
diff --git
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSink.java
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSink.java
index 59d8b4d..a01aa71 100644
---
a/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSink.java
+++
b/flink-connector-rocketmq/src/main/java/org/apache/flink/connector/rocketmq/sink/RocketMQSink.java
@@ -18,12 +18,6 @@
package org.apache.flink.connector.rocketmq.sink;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.JobInfo;
-import org.apache.flink.api.common.TaskInfo;
-import org.apache.flink.api.common.operators.MailboxExecutor;
-import org.apache.flink.api.common.operators.ProcessingTimeService;
-import org.apache.flink.api.common.serialization.SerializationSchema;
-import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.api.connector.sink2.CommitterInitContext;
import org.apache.flink.api.connector.sink2.Sink;
@@ -37,11 +31,7 @@ import
org.apache.flink.connector.rocketmq.sink.committer.SendCommittableSeriali
import org.apache.flink.connector.rocketmq.sink.writer.RocketMQWriter;
import
org.apache.flink.connector.rocketmq.sink.writer.serializer.RocketMQSerializationSchema;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
import
org.apache.flink.streaming.connectors.rocketmq.common.selector.MessageQueueSelector;
-import org.apache.flink.util.UserCodeClassLoader;
-
-import java.util.OptionalLong;
@PublicEvolving
public class RocketMQSink<IN> implements Sink<IN>,
SupportsCommitter<SendCommittable> {
@@ -75,77 +65,6 @@ public class RocketMQSink<IN> implements Sink<IN>,
SupportsCommitter<SendCommitt
configuration, messageQueueSelector, serializationSchema,
context);
}
- /**
- * Flink 1.x still declares the deprecated {@link Sink.InitContext}
overload abstract. The 1.x
- * runtime calls {@link #createWriter(WriterInitContext)}; this bridge
serves callers that pass
- * the old context type. Both interfaces expose the same methods.
- */
- @Override
- @Deprecated
- public SinkWriter<IN> createWriter(Sink.InitContext context) {
- return createWriter(new InitContextAdapter(context));
- }
-
- private static final class InitContextAdapter implements WriterInitContext
{
-
- private final Sink.InitContext delegate;
-
- private InitContextAdapter(Sink.InitContext delegate) {
- this.delegate = delegate;
- }
-
- @Override
- public UserCodeClassLoader getUserCodeClassLoader() {
- return delegate.getUserCodeClassLoader();
- }
-
- @Override
- public MailboxExecutor getMailboxExecutor() {
- return delegate.getMailboxExecutor();
- }
-
- @Override
- public ProcessingTimeService getProcessingTimeService() {
- return delegate.getProcessingTimeService();
- }
-
- @Override
- public SinkWriterMetricGroup metricGroup() {
- return delegate.metricGroup();
- }
-
- @Override
- public SerializationSchema.InitializationContext
- asSerializationSchemaInitializationContext() {
- return delegate.asSerializationSchemaInitializationContext();
- }
-
- @Override
- public boolean isObjectReuseEnabled() {
- return delegate.isObjectReuseEnabled();
- }
-
- @Override
- public <T> TypeSerializer<T> createInputSerializer() {
- return delegate.createInputSerializer();
- }
-
- @Override
- public OptionalLong getRestoredCheckpointId() {
- return delegate.getRestoredCheckpointId();
- }
-
- @Override
- public JobInfo getJobInfo() {
- return delegate.getJobInfo();
- }
-
- @Override
- public TaskInfo getTaskInfo() {
- return delegate.getTaskInfo();
- }
- }
-
@Override
public Committer<SendCommittable> createCommitter(CommitterInitContext
context) {
return new RocketMQCommitter(configuration);
diff --git a/pom.xml b/pom.xml
index 653aeb1..8fe589c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -81,11 +81,11 @@ under the License.
<properties>
<flink.parent.artifactId>flink-connector-rocketmq-parent</flink.parent.artifactId>
- <!-- Flink 1.x line targets Java 11 bytecode -->
- <target.java.version>11</target.java.version>
+ <!-- Flink 2.x requires Java 17 at compile time -->
+ <target.java.version>17</target.java.version>
<!-- Main Dependencies -->
- <flink.version>1.20.3</flink.version>
+ <flink.version>2.2.1</flink.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>
@@ -547,7 +547,7 @@ under the License.
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
- <release>11</release>
+ <release>17</release>
</configuration>
</plugin>
@@ -559,8 +559,8 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <!-- Same shade version as the 2.x branch: the one inherited
from
- flink-connector-parent bundles an ASM that cannot read
Java 17 class files. -->
+ <!-- The version inherited from flink-connector-parent bundles
an ASM that cannot
+ read Java 17 class files (major version 61), which Flink
2.x requires. -->
<version>3.6.0</version>
</plugin>
@@ -582,8 +582,11 @@ under the License.
<phase>verify</phase>
<configuration>
<!-- Upstream CI only runs `mvn clean package`, so
this verify-phase
- gate never ran there and its dependency list
is stale. Keep the
- report, but do not fail the build on it. -->
+ gate never ran there and its dependency list
is stale. Flink 2.x
+ additionally split flink-core / flink-runtime
/ flink-annotations
+ / flink-core-api into separate artifacts, so
the analyzer now
+ reports a long list of used-undeclared
provided-scope Flink
+ internals. Keep the report, but do not fail
the build on it. -->
<failOnWarning>false</failOnWarning>
<ignoreAllNonTestScoped>true</ignoreAllNonTestScoped>
</configuration>
