-
Notifications
You must be signed in to change notification settings - Fork 4.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[BEAM-1240] Create RabbitMqIO #1729
Conversation
Refer to this link for build results (access rights to CI server needed): |
Is this ready for review? Perhaps assign a reviewer? |
Yes. Just pinged Dan about that ;) |
@jbonofre is this PR still useful? If so, could you suggest a reviewer so this makes progress? |
@reuvenlax yes it is in complement of the AMQPIO. I will link with the PR I have on my own github (as requested by Eugene). |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks, a couple of questions.
- What benefit does this provide over using AmqpIO? RabbitMQ already implements the AMQP protocol, so AmqpIO can be used to talk to RabbitMQ, right?
- Both unit tests are currently marked "ignored", i.e. this code effectively doesn't have tests, and it will be easy for somebody to accidentally break it. What's the plan for dealing with this? I think this needs to be fixed before the PR can go in.
I will rebase and update this PR. About the questions:
Thanks ! |
Are you interested in getting some help to get the tests fixed? If so I might be able to help out. Writing embedded RabbitMQ tests can be quite tricky, but there might be a couple of workarounds:
WDYT? |
Sorry, I still don't understand why a |
Well that sounds reasonable indeed, unless there're some specific features of RabbitMQ that it cannot support?
To be honest I haven't even tried it, I just googled for Beam and rabbitmq support which lead me to this issue. Reading the amqp javadoc suggests that it only supports AMQP 1.0 (and for some reason I thought that was intentional), please correct me if I'm wrong. If it only supports AMQP 1.0, then I'm indeed blocked since we're using 0.9.1 in RabbitMQ. But then again, if we could make the AmqpIO support 0.9.1 then that sounds better to me. |
29c3b65
to
4b127a4
Compare
I fixed and updated Actually, it's the opposite in term of version: |
@jkff I fixed the IO and the tests. As said, AMQP IO (with usage of proton/qpid) only supports AMQP 1.0/1.1. In order to support AMQP 0.9.x (version used by RabbitMQ), we need to use the RabbitMQ client. That's the purpose of this PR (AMQP 0.9.x is very RabbitMQ specific). @johanhaleby the tests are now OK and the IO is ready to be tested. |
Is there no single AMQP client library for Java that supports both AMQP
1.0/1.1 and 0.9.x so that we could unify AmqpIO and this PR?
tis 7 nov. 2017 kl. 06:29 skrev Jean-Baptiste Onofré <
notifications@github.com>:
@jkff <https://github.com/jkff> I fixed the IO and the tests. As said,
AMQP IO (with usage of proton/qpid) only supports AMQP 1.0/1.1. In order to
support AMQP 0.9.x (version used by RabbitMQ), we need to use the RabbitMQ
client. That's the purpose of this PR (AMQP 0.9.x is very RabbitMQ
specific).
@johanhaleby <https://github.com/johanhaleby> the tests are now OK and
the IO is ready to be tested.
—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
<#1729 (comment)>, or mute
the thread
<https://github.com/notifications/unsubscribe-auth/AALalUVqMpXzuGKLuS6KlmajEsxkVdjnks5sz-q_gaJpZM4LZ0e4>
.
--
Sent from my phone
|
That's another option: I can gather both lib in As AMQP 0.9 is "only" RabbitMQ, I think it gives more visibility to have a dedicated IO. @jkff thoughts ? |
OK, I think I agree that RabbitMqIO makes sense for AMQP 0-9-1 because AFAIK RabbitMQ is the only system that implements AMQP 0-9-1 (all other "generic AMQP implementations" implement AMQP 1.0) so if people need it, it's probably because they are using RabbitMQ, and in that case they'll find it easier to locate it by RabbitMqIO. I'll take a look at the PR contents soon. |
|
||
@Override | ||
public void validate(PipelineOptions pipelineOptions) { | ||
checkState(connectionConfig() != null, "RabbitMqIO.read() requires a connection config " |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this should be in expand() because it doesn't use the PipelineOptions. Also should be rephrased:
checkState(..., "withConnectionConfig() is required.")
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
My bad, it's an "old" PR and I have to update to match the IO style guide. I'm doing it now, thanks !
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Actually, the checkState
doesn't make sense anymore as a default ConnectionConfig
is created in the constructor method. I remove this.
} | ||
|
||
@Override | ||
public Coder<byte[]> getDefaultOutputCoder() { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This method is deprecated, override getOutputCoder instead
} | ||
|
||
@Override | ||
public void validate() { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is a no-op, so this method can be omitted
// RabbitMQ uses queue, so, we can have several concurrent consumers as source | ||
List<RabbitMQSource> sources = new ArrayList<>(); | ||
for (int i = 0; i < desiredNumSplits; i++) { | ||
sources.add(new RabbitMQSource(spec)); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This can be simply sources.add(this)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is it not a new source for each split (to have concurrent consumers) ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
RabbitMQSource is simply an immutable java object, there's no difference between returning one object or returning another object that is completely equivalent to it. A Beam runner will anyway serialize them at construction time and deserialize each split on the respective worker.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Got it.
if (channel == null) { | ||
throw new IOException("No RabbitMQ channel available"); | ||
} | ||
channel.queueDeclare(source.spec.connectionConfig().queue(), false, false, false, null); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please annotate the false and null parameters with names for readability
if (channel == null) { | ||
throw new IOException("No RabbitMQ channel available"); | ||
} | ||
channel.queueDeclare(spec.connectionConfig().queue(), false, false, false, null); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm not sure what this code is trying to do. I believe RabbitMQ producers should publish to an exchange rather than a queue, and they shouldn't even care whether any queues bound to this exchange currently exist:
https://www.rabbitmq.com/tutorials/tutorial-three-python.html
The core idea in the messaging model in RabbitMQ is that the producer never sends any messages directly to a queue. Actually, quite often the producer doesn't even know if a message will be delivered to any queue at all. Instead, the producer can only send messages to an exchange....
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, forgot to merge that change. Merging.
broker = new Broker(); | ||
BrokerOptions options = new BrokerOptions(); | ||
options.setConfigProperty("qpid.amqp_port", String.valueOf(PORT)); | ||
options.setConfigProperty("qpid.work_dir", Files.createTempDirectory("qpid").toString()); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Use a @Rule TemporaryFolder
*/ | ||
public class RabbitMqIOTest { | ||
|
||
public static final int PORT = 5672; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What if this port is taken?
connectionFactory.setUri("amqp://guest:guest@localhost:" + PORT); | ||
Connection connection = connectionFactory.newConnection(); | ||
Channel channel = connection.createChannel(); | ||
channel.queueDeclare("READ", false, false, false, null); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Here too, the publisher should not be declaring queues, and rather it should be publishing to an exchange.
# specific language governing permissions and limitations | ||
# under the License. | ||
# | ||
guest:guest |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is it possible to specify this programmatically rather than via a configuration file?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, it should be do-able. I'm doing this.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Any progress?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ping?
219b390
to
76401c1
Compare
OK, seems that the changes are not done yet. Please ping this thread when ready for another round. |
76401c1
to
817c93d
Compare
SUCCESS --none-- |
@jkff I addressed most of your comments. Just checking to configuration the plain authentication provider programmatically in order to avoid the |
FAILURE --none-- |
SUCCESS --none-- |
retest this please |
SUCCESS --none-- |
@echauchot great thanks ! |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks! (my last review on this one - deferring to other reviewers afterwards)
* | ||
* <pre>{@code | ||
* pipeline.apply( | ||
* RabbitMqIO.read().withUri("amqp://user:password@localhost:5672").withQueue("QUEUE") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Missing ), and missing "PCollection messages = ..."
* | ||
* <h3>Publishing messages to RabbitMQ server</h3> | ||
* | ||
* <p>{@link RabbitMqIO} {@link Write} can send {@code byte[]} to a RabbitMQ server queue or |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's actually RabbitMqMessage
sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java
Show resolved
Hide resolved
sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java
Show resolved
Hide resolved
@Override | ||
public boolean advance() throws IOException { | ||
try { | ||
QueueingConsumer.Delivery delivery = consumer.nextDelivery(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Still relevant
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Show resolved
Hide resolved
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Outdated
Show resolved
Hide resolved
new Thread( | ||
() -> { | ||
try { | ||
Thread.sleep(5000); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Maybe a simple while with timeout to check the Qpid status to avoid flaky tests in case 5 sec is not enough for the QPid init time on an overloaded jenkins server.
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Outdated
Show resolved
Hide resolved
# specific language governing permissions and limitations | ||
# under the License. | ||
# | ||
guest:guest |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ping?
@jkff thanks for the update, I will address your comments ! |
Hi All Thanks for RmqIO :) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Show resolved
Hide resolved
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Outdated
Show resolved
Hide resolved
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Show resolved
Hide resolved
sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java
Show resolved
Hide resolved
5cac406
to
f1665f4
Compare
I exposed |
retest this please |
Thanks @jbonofre |
Hello Folks While working with RmqIO and Flink local runner of the beam, I observed, Even if Rmq has data to consume, Beam pipeline Job execution switched to status FINISHED without any exception. Code : It will be great if someone can help me out about this random behavior. |
f1665f4
to
ca90832
Compare
@akankshajain18 that's probably related to the watermark behavior which is not correct for now. I'm fixing that and other comments. I'm moving forward on this PR. Sorry for the delay. |
424bea6
to
84c948e
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks JB !
minor changes on historical comments + my first round comments and we are done !
bfa77c6
to
a4d9e5b
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for addressing my comments JB! You just missed the one on equals impl and the one on Thread.sleep for which I added 2 suggestions. Also suggested to use maxNumRecords in place of absolute 10 in one of the tests.
sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqMessage.java
Outdated
Show resolved
Hide resolved
p.run(); | ||
|
||
while (received.size() < maxNumRecords) { | ||
Thread.sleep(500); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
same here
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Outdated
Show resolved
Hide resolved
sdks/java/io/rabbitmq/src/test/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIOTest.java
Show resolved
Hide resolved
p.run(); | ||
|
||
while (received.size() < maxNumRecords) { | ||
Thread.sleep(500); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There is no problem with this sleep because it is not a wait for an absolute period of time but more a way to temporise the check of the receive queue depth. As a very minor suggestion (only if you feel the envy :) ), I would implement it with a new CountDownLatch(maxNumRecords); countdown.await() and countdown.countDown in the TestConsumer when it receives a message.
But once again it is very minor way to reduce test time a tiny bit but I'm ok with current impl
8a205ac
to
ebfe51b
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM ! Thanks for your work JB ! Please squash the commits into the first one and ping me when it's done so that I can merge the PR
0577b86
to
119d0bc
Compare
@echauchot Squash & rebase done. |
I tried to use the
After investigation it looks like it's caused by the fact that no default is given to
It looks likes this bug as already been raised on Jira and on Stack Overflow Do you think it's a problem that needs to be corrected ? Is there any workaround you might think of ? |
@alafanechere agree, can you please create a dedicated Jira about that ? I will fix it. |
@alafanechere by the way, I can create the Jira for you (and I will assign the Jira to me). Please let me know. |
Thanks ! I juste created the JIRA : https://jira.apache.org/jira/browse/BEAM-6424 |
Be sure to do all of the following to help us incorporate your contribution
quickly and easily:
[BEAM-<Jira issue #>] Description of pull request
mvn clean verify
. (Even better, enableTravis-CI on your fork and ensure the whole test matrix passes).
<Jira issue #>
in the title with the actual Jira issuenumber, if there is one.
Individual Contributor License Agreement.