[FLINK-20628] RabbitMQ Connector using FLIP-27 Source API #1

pscls · 2022-05-23T18:36:33Z

What is the purpose of the change

This pull request ports the RabbitMQ connector implementation to the new Connector’s API described in FLIP-27 and FLIP-143. It includes both source and sink with at-most-once, at-least-once, and exactly-once behavior, respectively.

This pull request closes the following issues (separated RabbitMQ connector Source and Sink tickets): FLINK-20628 and FLINK-21373

Brief change log

Source and Sink use the RabbitMQ’s Java Client API to interact with RabbitMQ
The RabbitMQ Source reads messages from a queue
At-least-once
- Messages are acknowledged on checkpoint completion
Exactly-once
- Messages are acknowledged in a transaction
The user has to set correlation ids for deduplication
The RabbitMQ Sink publishes messages to a queue
At-least-once
- Unacknowledged messages are resend on checkpoints
Exactly-once
- Messages between two checkpoints are published in a transaction

Verifying this change

This change added tests and can be verified as follows:

All changes are within the flink-connectors/flink-connector-rabbitmq2/ module.
Added Integration Tests can be find under org.apache.flink.connector.rabbitmq2.source and org.apache.flink.connector.rabbitmq2.sink package in the test respective directories.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (don't know)
The runtime per-record code paths (performance sensitive): (don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (don't know)
The S3 file system connector: (no)

Documentation

Does this pull request introduces a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)

Co-authored-by: Yannik Schroeder [email protected]
Co-authored-by: Jan Westphal [email protected]

pscls · 2022-05-23T18:37:28Z

This is a copy from the original PR (apache/flink#15140) against the Flink repository.

pscls · 2022-05-23T18:41:03Z

@MartijnVisser We are not exactly sure what has to be part of the root-pom.

MartijnVisser · 2022-05-24T12:38:01Z

@pscls I think you've done a good job already with the root-pom; it looks like the one we currently have for Elasticsearch. I've just approved the run, so we can also see how the build behaves. When I tried it locally, it complained about https://github.com/pscls/flink-connector-rabbitmq/blob/new-api-connector/flink-connector-rabbitmq/src/main/java/org/apache/flink/connector/rabbitmq/sink/RabbitMQSink.java#L124 having a whitespace, but now the tests are running for me.

I'll work on finding someone who can help with the review for this.

MartijnVisser · 2022-06-20T07:32:54Z

@pscls Can you have a look at the failing build? It's a checkstyle error.

wanglijie95 · 2022-06-20T07:43:07Z

@MartijnVisser @pscls I noticed that the GitBox of flink-connector-rabbitmq sent emails to the [email protected]. Is it expected?

MartijnVisser · 2022-06-20T07:56:25Z

@wanglijie95 No. Most likely this is caused because the PR was created/is not yet using the ASF config as defined in https://github.com/apache/flink-connector-rabbitmq/blob/main/.asf.yaml

MartijnVisser · 2022-08-03T13:06:17Z

@pscls The CI fails due to spotless; can you fix that? (By running mvn spotless:apply)

pscls · 2022-08-09T08:16:47Z

@pscls The CI fails due to spotless; can you fix that? (By running mvn spotless:apply)

@MartijnVisser I've nothing to commit when running mvn spotless:apply.

MartijnVisser · 2022-08-31T11:53:55Z

@pscls Weird. Could you push once more, since the logs are no longer available?

RabbitMQ Connector using the new Source API https://issues.apache.org/jira/browse/FLINK-20628 Co-authored-by: Yannik Schroeder <[email protected]> Co-authored-by: Jan Westphal <[email protected]>

RabbitMQ Connector using the new Sink API https://issues.apache.org/jira/browse/FLINK-21373 Co-authored-by: Yannik Schroeder <[email protected]> Co-authored-by: Jan Westphal <[email protected]>

zentol · 2022-09-29T10:26:19Z

flink-connector-rabbitmq/src/main/java/org/apache/flink/connector/rabbitmq/sink/README.md

+Users that create and publish derivative work based on Flink's
+RabbitMQ connector (thereby re-distributing the "RabbitMQ AMQP Java Client")
+must be aware that this may be subject to conditions declared in the
+Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL")
+and the Apache License version 2 ("ASL").


This belongs into a NOTICE, both contained in the source release and jars.

Suggested change

Users that create and publish derivative work based on Flink's

RabbitMQ connector (thereby re-distributing the "RabbitMQ AMQP Java Client")

must be aware that this may be subject to conditions declared in the

Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL")

and the Apache License version 2 ("ASL").

Users that create and publish derivative work based on Flink's

RabbitMQ connector (thereby re-distributing the "RabbitMQ AMQP Java Client")

must be aware that this is subject to conditions declared in either the

Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2 ("GPL")

OR the Apache License version 2 ("ASL").

Additionally this needs clarification that users can choose which license they use for the derivative work.

zentol · 2022-09-29T10:27:25Z

pom.xml

+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>org.apache.flink</groupId>
+	<artifactId>flink-connectors</artifactId>


This file is missing a fair amount of cleanups that were later applied to the ES parent pom.

See apache/flink-connector-elasticsearch#31 and replicate the changes.

zentol · 2022-09-29T10:29:00Z

flink-connector-rabbitmq/pom.xml

+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-jar-plugin</artifactId>
+				<executions>
+					<execution>
+						<goals>
+							<goal>test-jar</goal>
+						</goals>
+					</execution>
+				</executions>
+			</plugin>
+		</plugins>
+	</build>


Why are we publishing a test-jar?

zentol · 2022-09-29T10:30:10Z

...java/org/apache/flink/connector/rabbitmq/source/split/RabbitMQSourceSplitSerializerTest.java

+
+import org.apache.flink.connector.rabbitmq.common.RabbitMQConnectionConfig;
+
+import org.junit.Test;


It would be good if we could start out with junit5 and not have to go through a migration again.

zentol · 2022-09-29T10:30:32Z

flink-connector-rabbitmq/README.md

+from a RabbitMQ queue in three different consistency modes: at-most-once, at-least-once,
+and exactly-once.
+
+## Consistency Modes


This all belongs into the documentation, not a readme.

zentol · 2022-09-29T10:58:43Z

...rc/main/java/org/apache/flink/connector/rabbitmq/source/reader/RabbitMQSourceReaderBase.java

+        return CompletableFuture.runAsync(
+                () -> {
+                    while (!collector.hasUnpolledMessages()) {
+                        // supposed to be empty
+                    }
+                });


This is problematic for a few reasons.

a) It runs in a non-specified thread pool, which means it runs in the JVMs common pool which may also be in use by other components. Use a dedicated executor.
b) It hot-loops, which both blocks an entire thread from doing anything and blows through CPU cycles. Consider restructuring the collector to return a sort of availability future that is completed once a message was added, or use basic locking to at least prevent hot-looping.

zentol · 2022-09-29T10:59:09Z

...rc/main/java/org/apache/flink/connector/rabbitmq/source/reader/RabbitMQSourceReaderBase.java

+
+    @Override
+    public List<RabbitMQSourceSplit> snapshotState(long checkpointId) {
+        return split != null ? Collections.singletonList(split.copy()) : new ArrayList<>();


Suggested change

return split != null ? Collections.singletonList(split.copy()) : new ArrayList<>();

return split != null ? Collections.singletonList(split.copy()) : Collections.emptyList();

zentol · 2022-09-29T10:59:30Z

...rc/main/java/org/apache/flink/connector/rabbitmq/source/reader/RabbitMQSourceReaderBase.java

+        if (split != null) {
+            return;
+        }


This seems wrong.

zentol · 2022-09-29T11:00:34Z

...nector-rabbitmq/src/main/java/org/apache/flink/connector/rabbitmq/source/RabbitMQSource.java

+         */
+        public RabbitMQSourceBuilder<T> setDeserializationSchema(
+                DeserializationSchema<T> deserializationSchema) {
+            this.deserializationSchema = deserializationSchema;


missing null checks; preferable to do them here to fail as early as possible.

zentol · 2022-09-29T11:02:00Z

...ctor-rabbitmq/src/test/java/org/apache/flink/connector/rabbitmq/common/RabbitMQBaseTest.java

+ * provides behavior to easily add onto the stream, send message to RabbitMQ and get the messages in
+ * RabbitMQ.
+ */
+public abstract class RabbitMQBaseTest {


Please don't use test bases; they've always ended up creating problems.

Model this as an extension instead.

zentol · 2022-09-30T09:34:48Z

You can avoid a lot of boilerplate by using the preliminary flink-connector-parent pom as shown here: apache/flink-connector-elasticsearch@6e30d5d

RocMarshal · 2023-06-12T10:10:03Z

Hi, @pscls Thank you very much for the contribution.
I notice that this PR has not been updated for a long time.
Would you like to continue advancing it ?
After the PR completed, FLINK-25380 will be introduced.
Looking forward to your opinion.
Thanks.

MartijnVisser · 2023-06-29T14:27:42Z

Would you like to continue advancing it ?

@RocMarshal Do you want to take this over?

RocMarshal · 2023-06-29T15:22:12Z

Would you like to continue advancing it ?

@RocMarshal Do you want to take this over?

Hi, @MartijnVisser Glad to get your attention.
In fact, you have already assigned this ticket to me on https://issues.apache.org/jira/browse/FLINK-20628 . I am working for it now.
Thank you ~

MartijnVisser · 2023-07-28T11:12:45Z

I am working for it now.

Any update to report from your end on this @RocMarshal ?

RocMarshal · 2023-07-29T04:14:15Z

I am working for it now.

Any update to report from your end on this @RocMarshal ?

Hi, @MartijnVisser Still in doing. I'll update in the end of this week.

pscls mentioned this pull request May 23, 2022

[FLINK-20628][connectors/rabbitmq2] RabbitMQ connector using new connector API apache/flink#15140

Closed

pscls force-pushed the new-api-connector branch from 84c341f to c739b58 Compare June 20, 2022 10:17

westphal-jan and others added 2 commits September 12, 2022 18:37

[FLINK-20628] RabbitMQ Connector using FLIP-27 Source API

1fe3f7e

RabbitMQ Connector using the new Source API https://issues.apache.org/jira/browse/FLINK-20628 Co-authored-by: Yannik Schroeder <[email protected]> Co-authored-by: Jan Westphal <[email protected]>

[FLINK-21373] RabbitMQ Connector using FLIP-143 Sink API

77cd700

RabbitMQ Connector using the new Sink API https://issues.apache.org/jira/browse/FLINK-21373 Co-authored-by: Yannik Schroeder <[email protected]> Co-authored-by: Jan Westphal <[email protected]>

pscls force-pushed the new-api-connector branch from c739b58 to 77cd700 Compare September 12, 2022 11:38

MartijnVisser mentioned this pull request Sep 29, 2022

[FLINK-29467][CI] Update CI workflow #2

Merged

zentol requested changes Sep 29, 2022

View reviewed changes

zentol self-assigned this Sep 29, 2022

RocMarshal mentioned this pull request Sep 12, 2023

[FLINK-20628] Port RabbitMQ Connector using unified Source #16

Closed

zentol removed their assignment Mar 4, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-20628] RabbitMQ Connector using FLIP-27 Source API #1

[FLINK-20628] RabbitMQ Connector using FLIP-27 Source API #1

pscls commented May 23, 2022

pscls commented May 23, 2022

pscls commented May 23, 2022

MartijnVisser commented May 24, 2022

MartijnVisser commented Jun 20, 2022

wanglijie95 commented Jun 20, 2022

MartijnVisser commented Jun 20, 2022

MartijnVisser commented Aug 3, 2022

pscls commented Aug 9, 2022

MartijnVisser commented Aug 31, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol Sep 29, 2022

zentol commented Sep 30, 2022

RocMarshal commented Jun 12, 2023

MartijnVisser commented Jun 29, 2023

RocMarshal commented Jun 29, 2023

MartijnVisser commented Jul 28, 2023

RocMarshal commented Jul 29, 2023


		import org.apache.flink.connector.rabbitmq.common.RabbitMQConnectionConfig;

		import org.junit.Test;

	return split != null ? Collections.singletonList(split.copy()) : new ArrayList<>();
	return split != null ? Collections.singletonList(split.copy()) : Collections.emptyList();

[FLINK-20628] RabbitMQ Connector using FLIP-27 Source API #1

Are you sure you want to change the base?

[FLINK-20628] RabbitMQ Connector using FLIP-27 Source API #1

Conversation

pscls commented May 23, 2022

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

pscls commented May 23, 2022

pscls commented May 23, 2022

MartijnVisser commented May 24, 2022

MartijnVisser commented Jun 20, 2022

wanglijie95 commented Jun 20, 2022

MartijnVisser commented Jun 20, 2022

MartijnVisser commented Aug 3, 2022

pscls commented Aug 9, 2022

MartijnVisser commented Aug 31, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zentol commented Sep 30, 2022

RocMarshal commented Jun 12, 2023

MartijnVisser commented Jun 29, 2023

RocMarshal commented Jun 29, 2023

MartijnVisser commented Jul 28, 2023

RocMarshal commented Jul 29, 2023