You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/06/20 19:10:58 UTC
[GitHub] [flink] austince opened a new pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
austince opened a new pull request #12729:
URL: https://github.com/apache/flink/pull/12729
## What is the purpose of the change
This PR allows configuring the RabbitMQ source's prefetch count, allowing jobs to properly handle backpressure and fix the OOM issue that arises when too many messages are sent from the RMQ server to handle. This is left opt-in and shouldn't affect any current jobs if unset.
FLINK-10195 contains the discussion around the tradeoffs and links to relevant docs/ proofs-of-concept. This takes some guidance from https://github.com/apache/flink/pull/8111, though just pulls in the RMQ QueueingConsumer and sets the channel prefetch instead of starting/ stopping the consumer. I have not updated the dependency to the latest version but have set us up to do so easily. Should I open up a new ticket for that, or just do it in this PR?
## Brief change log
- adds the ability to `setPrefetchCount` on the `RMQConnectionConfig` and its builder
- breaks out creating the source's `Channel` into a protected method and configures the prefetch count, if set
- moves the `com.rabbitmq.client.QueueingConsumer` into Flink, as it is removed in later versions
- adds licensing information for the `com.rabbitmq` package
## Verifying this change
This change added tests and can be verified as follows:
- added `RMQSourceTest#testSetPrefetchCount`, which tests setting a prefetch count creates the channel with it
- added `RMQSinkTest#testUnsetPrefetchCount`, which tests leaving the prefetch count unset creates the channel as it normally does
- added `RMQConnectionConfigTest#shouldReturnEmptyOptionalPrefetchCount`, which tests the returned `Optional` is empty when unset
- added `RMQConnectionConfigTest#shouldReturnEmptyOptionalPrefetchCount`, which tests the returned `Optional` is present when set
-- I would love to add an e2e test for this but unsure how to go about it. Would that best be done in another ticket?
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / **docs** / JavaDocs / not documented)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 99231d5b11dd2adced76b48996dfb252b5cc9fc2 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ce89d0b66d61398909e82fd603181f4976351f48 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693)
* 765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-662736061
@dawidwys @aljoscha - hey guys, are you able to take a look at this when you have a chance? Thank you!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-662469562
@flinkbot run azure
Had to update a test for the global flag addition, but all other tests are passing.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] senegalo commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
senegalo commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-650744581
@austince Really nice Awesome work .. i too need this badly in one of our production systems as we always run into memory issues when we restart the app and have to bump it to ridiculous values when restarting the app !
Had just the one tiny comment regarding the `global` value in the `basicQos` method.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-662018668
@flinkbot run azure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 99231d5b11dd2adced76b48996dfb252b5cc9fc2 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ce89d0b66d61398909e82fd603181f4976351f48 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693)
* 765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] senegalo commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
senegalo commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-648725837
i promise i will look into it this weekend ... i am sorry i was a bit sick lately and couldn't focus on anything :(
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on a change in pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on a change in pull request #12729:
URL: https://github.com/apache/flink/pull/12729#discussion_r458283453
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
##########
@@ -141,6 +140,22 @@ protected Connection setupConnection() throws Exception {
return setupConnectionFactory().newConnection();
}
+ /**
+ * Initializes the consumer's {@link Channel}. If a prefetch count has been set in {@link RMQConnectionConfig},
+ * the new channel will be use it for {@link Channel#basicQos(int)}.
+ *
+ * @param connection the consumer's {@link Connection}.
+ * @return the channel.
+ * @throws Exception if there is an issue creating or configuring the channel.
+ */
+ protected Channel setupChannel(Connection connection) throws Exception {
+ Channel chan = connection.createChannel();
+ if (rmqConnectionConfig.getPrefetchCount().isPresent()) {
+ chan.basicQos(rmqConnectionConfig.getPrefetchCount().get());
Review comment:
Finally got around to this, sorry about the late reply! When I was putting together the example, I found that RMQ 3.3 is not even available on Docker hub and has [been out of service since 2015](https://www.rabbitmq.com/versions.html) and [the last patch was released in 2016](https://github.com/rabbitmq/rabbitmq-server/releases/tag/rabbitmq_v3_3_5), so I think we can safely just set the global flag and not care about that version. Here's the playground running this branch's code: https://github.com/austince/flink-rmq-playground
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647060826
Build is failing because the two files moved in from the RMQ source -- should I just replace that license header with the normal one? Should I keep the list of changes somewhere?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on a change in pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on a change in pull request #12729:
URL: https://github.com/apache/flink/pull/12729#discussion_r446980990
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
##########
@@ -141,6 +140,22 @@ protected Connection setupConnection() throws Exception {
return setupConnectionFactory().newConnection();
}
+ /**
+ * Initializes the consumer's {@link Channel}. If a prefetch count has been set in {@link RMQConnectionConfig},
+ * the new channel will be use it for {@link Channel#basicQos(int)}.
+ *
+ * @param connection the consumer's {@link Connection}.
+ * @return the channel.
+ * @throws Exception if there is an issue creating or configuring the channel.
+ */
+ protected Channel setupChannel(Connection connection) throws Exception {
+ Channel chan = connection.createChannel();
+ if (rmqConnectionConfig.getPrefetchCount().isPresent()) {
+ chan.basicQos(rmqConnectionConfig.getPrefetchCount().get());
Review comment:
Thanks for going through this! That's a good point, I'm not sure what a good answer is. I've read in the Nodejs client I used that [sending that flag to a RMQ cluster < version 3.3.0 will kill it entirely](https://www.squaremobius.net/amqp.node/channel_api.html#channel_prefetch), but I haven't tested that out at all or seen it elsewhere. I'll put together a playground this week.
We're always reading off of one queue per source in our jobs, but our use case is pretty simple so I'm not sure I've got the best understanding of other RMQ users. What kinds of reasons do people have for sharing a channel?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647035012
@senegalo, would you let me know what you think of this if you have time? Thank you!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 790ebc88d93408cba281d3836ea24e6f6bc4a866 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696)
* 790ebc88d93408cba281d3836ea24e6f6bc4a866 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4809",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 790ebc88d93408cba281d3836ea24e6f6bc4a866 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4809) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810) Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-665680169
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] senegalo commented on a change in pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
senegalo commented on a change in pull request #12729:
URL: https://github.com/apache/flink/pull/12729#discussion_r447089528
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
##########
@@ -141,6 +140,22 @@ protected Connection setupConnection() throws Exception {
return setupConnectionFactory().newConnection();
}
+ /**
+ * Initializes the consumer's {@link Channel}. If a prefetch count has been set in {@link RMQConnectionConfig},
+ * the new channel will be use it for {@link Channel#basicQos(int)}.
+ *
+ * @param connection the consumer's {@link Connection}.
+ * @return the channel.
+ * @throws Exception if there is an issue creating or configuring the channel.
+ */
+ protected Channel setupChannel(Connection connection) throws Exception {
+ Channel chan = connection.createChannel();
+ if (rmqConnectionConfig.getPrefetchCount().isPresent()) {
+ chan.basicQos(rmqConnectionConfig.getPrefetchCount().get());
Review comment:
Hemm ... you're absolutely right i also have a very basic setup several independent brokers and we duplicate and deduplicate on the clients.
`To keep life interesting, using the global flag with an RabbitMQ older than v3.3.0 will bring down the whole connection.` :rofl: okay then experimenting is the way to go ! do you need a helping hand ? i can spare a couple of hours next weekend if you need help.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696)
* 790ebc88d93408cba281d3836ea24e6f6bc4a866 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ce89d0b66d61398909e82fd603181f4976351f48 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647034977
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 99231d5b11dd2adced76b48996dfb252b5cc9fc2 (Sat Jun 20 19:12:52 UTC 2020)
**Warnings:**
* Documentation files were touched, but no `.zh.md` files: Update Chinese documentation or file Jira ticket.
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 99231d5b11dd2adced76b48996dfb252b5cc9fc2 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864)
* ce89d0b66d61398909e82fd603181f4976351f48 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-662015553
@flinkbot run azure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647035087
> **Warnings:**
>
> * Documentation files were touched, but no `.zh.md` files: Update Chinese documentation or file Jira ticket.
>
Should I open a ticket before this is finalized, or after?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647035129
@flinkbot run azure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4809",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 790ebc88d93408cba281d3836ea24e6f6bc4a866 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4809) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince removed a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince removed a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-662015553
@flinkbot run azure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys merged pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
dawidwys merged pull request #12729:
URL: https://github.com/apache/flink/pull/12729
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
dawidwys edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-665737595
You can create a separate commit in this PR, where you do the bump and remove the `Delivery` class. You can tag this commit with FLINK-17529.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on a change in pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on a change in pull request #12729:
URL: https://github.com/apache/flink/pull/12729#discussion_r462315294
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/resources/META-INF/NOTICE
##########
@@ -0,0 +1,9 @@
+flink-connector-rabbitmq
+Copyright 2014-2020 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
+
+- com.rabbitmq:amqp-client:4.2.0
Review comment:
Ok great, thank you! Are the changes I listed in the source files themselves good enough for ASL license-compliance? Never modified source into a project of this scale before...
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
##########
@@ -141,6 +140,25 @@ protected Connection setupConnection() throws Exception {
return setupConnectionFactory().newConnection();
}
+ /**
+ * Initializes the consumer's {@link Channel}. If a prefetch count has been set in {@link RMQConnectionConfig},
+ * the new channel will be use it for {@link Channel#basicQos(int)}.
+ *
+ * @param connection the consumer's {@link Connection}.
+ * @return the channel.
+ * @throws Exception if there is an issue creating or configuring the channel.
+ */
+ protected Channel setupChannel(Connection connection) throws Exception {
Review comment:
I think both are great ideas, I'll implement those shortly.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-665660452
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-665735267
> What about the version bump? Do you want to do it as part of this PR?
Yeah, sorry to miss that note, I'm good to do that as well! Should I tag the FLINK-17529 ticket as well?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #12729:
URL: https://github.com/apache/flink/pull/12729#discussion_r462286847
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/resources/META-INF/NOTICE
##########
@@ -0,0 +1,9 @@
+flink-connector-rabbitmq
+Copyright 2014-2020 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
+
+- com.rabbitmq:amqp-client:4.2.0
Review comment:
We don't need the file for this dependency for two reasons:
1. We do not bundle the dependency. We need to create the NOTICE file if we e.g. shade the dependency into an uber-jar. In this case user has to manually add/bundle the dependency for the consumer to work.
2. It is ASL license.
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
##########
@@ -141,6 +140,25 @@ protected Connection setupConnection() throws Exception {
return setupConnectionFactory().newConnection();
}
+ /**
+ * Initializes the consumer's {@link Channel}. If a prefetch count has been set in {@link RMQConnectionConfig},
+ * the new channel will be use it for {@link Channel#basicQos(int)}.
+ *
+ * @param connection the consumer's {@link Connection}.
+ * @return the channel.
+ * @throws Exception if there is an issue creating or configuring the channel.
+ */
+ protected Channel setupChannel(Connection connection) throws Exception {
Review comment:
If I understand the RMQSource well, the reasons why `setupConnection` or `setupConnectionFactory` are protected is for assertions in tests. I see you managed to test it without using the `setupChannel` method. I think we can safely make it private.
Moreover WDYT of annotating the other methods with `VisibleForTesting`?
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/resources/META-INF/NOTICE
##########
@@ -0,0 +1,9 @@
+flink-connector-rabbitmq
+Copyright 2014-2020 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software License 2.0. (http://www.apache.org/licenses/LICENSE-2.0.txt)
+
+- com.rabbitmq:amqp-client:4.2.0
Review comment:
Yes, the classes itself are good, imo.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4809",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
}, {
"hash" : "491460bd0b5a9b2e67170d32ca2ffc3506950f65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=5012",
"triggerID" : "665680169",
"triggerType" : "MANUAL"
}, {
"hash" : "491460bd0b5a9b2e67170d32ca2ffc3506950f65",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=5012",
"triggerID" : "491460bd0b5a9b2e67170d32ca2ffc3506950f65",
"triggerType" : "PUSH"
}, {
"hash" : "5ffd02df057f21695027531e82e0ec717c2ddb91",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=5019",
"triggerID" : "5ffd02df057f21695027531e82e0ec717c2ddb91",
"triggerType" : "PUSH"
}, {
"hash" : "",
"status" : "DELETED",
"url" : "TBD",
"triggerID" : "665680169",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 5ffd02df057f21695027531e82e0ec717c2ddb91 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=5019)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] austince commented on a change in pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
austince commented on a change in pull request #12729:
URL: https://github.com/apache/flink/pull/12729#discussion_r447320484
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
##########
@@ -141,6 +140,22 @@ protected Connection setupConnection() throws Exception {
return setupConnectionFactory().newConnection();
}
+ /**
+ * Initializes the consumer's {@link Channel}. If a prefetch count has been set in {@link RMQConnectionConfig},
+ * the new channel will be use it for {@link Channel#basicQos(int)}.
+ *
+ * @param connection the consumer's {@link Connection}.
+ * @return the channel.
+ * @throws Exception if there is an issue creating or configuring the channel.
+ */
+ protected Channel setupChannel(Connection connection) throws Exception {
+ Channel chan = connection.createChannel();
+ if (rmqConnectionConfig.getPrefetchCount().isPresent()) {
+ chan.basicQos(rmqConnectionConfig.getPrefetchCount().get());
Review comment:
I'll let you know how it goes and thank you for the offer - I'll take you up on it if I can't find time this week! Moving this weekend, so it'll be busy over here..
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12729:
URL: https://github.com/apache/flink/pull/12729#issuecomment-647037124
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=3864",
"triggerID" : "99231d5b11dd2adced76b48996dfb252b5cc9fc2",
"triggerType" : "PUSH"
}, {
"hash" : "ce89d0b66d61398909e82fd603181f4976351f48",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4693",
"triggerID" : "ce89d0b66d61398909e82fd603181f4976351f48",
"triggerType" : "PUSH"
}, {
"hash" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4696",
"triggerID" : "765fcb8c6c01c00bb15a3ce256e2608dc1ffe7ca",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"triggerType" : "PUSH"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4809",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "647035129",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "662018668",
"triggerType" : "MANUAL"
}, {
"hash" : "790ebc88d93408cba281d3836ea24e6f6bc4a866",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808",
"triggerID" : "662469562",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 790ebc88d93408cba281d3836ea24e6f6bc4a866 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4809) Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4739) Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4810) Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=4808)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] senegalo commented on a change in pull request #12729: [FLINK-10195][connectors/rabbitmq] Allow setting QoS
Posted by GitBox <gi...@apache.org>.
senegalo commented on a change in pull request #12729:
URL: https://github.com/apache/flink/pull/12729#discussion_r446643314
##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSource.java
##########
@@ -141,6 +140,22 @@ protected Connection setupConnection() throws Exception {
return setupConnectionFactory().newConnection();
}
+ /**
+ * Initializes the consumer's {@link Channel}. If a prefetch count has been set in {@link RMQConnectionConfig},
+ * the new channel will be use it for {@link Channel#basicQos(int)}.
+ *
+ * @param connection the consumer's {@link Connection}.
+ * @return the channel.
+ * @throws Exception if there is an issue creating or configuring the channel.
+ */
+ protected Channel setupChannel(Connection connection) throws Exception {
+ Channel chan = connection.createChannel();
+ if (rmqConnectionConfig.getPrefetchCount().isPresent()) {
+ chan.basicQos(rmqConnectionConfig.getPrefetchCount().get());
Review comment:
should we set the global flag to true ? to enforce having only that limit throughout the channel ?
i am worried if someone adds a functionality that say requires having multiple consumers on the same channel that then the `preFetchCount` would be per consumer and the preFetchCount set by the client would suddenly mean a completely different thing. But if enforced here to be per channel it will have the same effect for the user.
According to the documentation `chan.basicQos(prefectchCount, boolean global)` and the boolean `global` is:
```
false | applied separately to each new consumer on the channel
true | shared across all consumers on the channel
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org