GitHub user pduveau opened a pull request:
https://github.com/apache/flink/pull/5410
[FLINK-8468] [RabbitMQ Connector] Take advantage of AMQP features
## The purpose of the change
To extend the Sink Connector to take full advantage of AMQP features. The
AMQP Message sent can :
- receive headers computed using the incoming data,
- be published to a computed exchange,
- be routed with a computed routing key.
This is done while using a new constructor. The existing constructor was
not modified for compatibility but the queue creation on setup can be disabled
by configuration.
## Change log
A interface class is added to define methods to compute the three different
parameter listed above.
A new constructor is added to the Sink class that takes an implemented
object of this interface as parameter.
Using this new constructor also disable the queue creation (queueName
attribute is null)
## Verifying this change
This change added tests and can be verified as follows:
- a new test class build on existing one is provided
- the test cover the new mechanisms for the three publishing parameter.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/pduveau/flink Extended_RabbitMQ_Connector
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/5410.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #5410
----
commit 64072b1c8e6cc0af3203ffff61bd518ca4b0ed50
Author: Philippe Duveau <philippe.duveau@...>
Date: 2018-02-04T00:15:25Z
[FLINK-8468] Make the connector to take advantage of AMQP features (routing
key, exchange and message properties)
commit 388278eb5ed8e8330d6c23555d81d87bc12dc8fd
Author: Philippe Duveau <philippe.duveau@...>
Date: 2018-02-04T10:10:04Z
[FLINK-8468] Make the connector to take advantage of AMQP features (routing
key, exchange and message properties) and Test
commit eec3504c29e1f48d535a2fc65423fa5ccfe8985e
Author: Philippe Duveau <philippe.duveau@...>
Date: 2018-02-04T10:28:44Z
[FLINK-8468] Make the connector to take advantage of AMQP features (routing
key, exchange and message properties) and Test
commit c900f507690ddaa7ec9734e5002fe1bc2494c5da
Author: Philippe Duveau <philippe.duveau@...>
Date: 2018-02-04T10:37:53Z
[FLINK-8468] Make the connector to take advantage of AMQP features (routing
key, exchange and message properties) and Test
----
---
