You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "fabrizziocht (via GitHub)" <gi...@apache.org> on 2024/12/04 06:33:43 UTC
[D] Is RabbitMQ source connectors safe? [pulsar]
GitHub user fabrizziocht created a discussion: Is RabbitMQ source connectors safe?
I was exploring the rabbitMQ source connector (https://github.com/apache/pulsar/blob/v4.0.1/pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java) that implements the class PushSource (it extends AbstractPushSource class),
I can see that each time a message is collected by the RabbitMQConsumer class using the handleDelivery method it invokes the RabbitMQSource consume:
https://github.com/apache/pulsar/blob/v4.0.1/pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java#L99
Then it stores only in memory in a LinkedBlockingQueue and the ack is sent to RabbitMQ:
https://github.com/apache/pulsar/blob/master/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/AbstractPushSource.java#L76
What happens if the pulsar IO service is restarted? I can see that the data stored in the LinkedBlockingQueue will be lost, because the message was already acked and removed in rabbitMQ and also was not still written in the pulsar topic. Is it a limitation of the rabbitmq source connector? or all connectors that implements the PushSource has the problem?
GitHub link: https://github.com/apache/pulsar/discussions/23677
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
Re: [D] Is RabbitMQ source connectors safe? [pulsar]
Posted by "fabrizziocht (via GitHub)" <gi...@apache.org>.
GitHub user fabrizziocht edited a discussion: Is RabbitMQ source connectors safe?
I was exploring the rabbitMQ source connector (https://github.com/apache/pulsar/blob/v4.0.1/pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java) that implements the class PushSource (it extends AbstractPushSource class),
I can see that each time a message is collected by the RabbitMQConsumer class using the handleDelivery method it invokes the RabbitMQSource consume:
https://github.com/apache/pulsar/blob/v4.0.1/pulsar-io/rabbitmq/src/main/java/org/apache/pulsar/io/rabbitmq/RabbitMQSource.java#L99
Then it stores only in memory in a LinkedBlockingQueue and the ack is sent to RabbitMQ:
https://github.com/apache/pulsar/blob/master/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/AbstractPushSource.java#L76
What happens if the pulsar IO service is restarted? I can see that the data stored in the LinkedBlockingQueue will be lost, because the message was already acked and removed in rabbitMQ and also was not still written in the pulsar topic. Is it a limitation of the rabbitmq source connector? or all connectors that implements the PushSource has the problem? Exists another interface/class that can guarantee that the message was written first in the pulsar topic before send the ack to rabbitMQ?
GitHub link: https://github.com/apache/pulsar/discussions/23677
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
Re: [D] Is RabbitMQ source connectors safe? [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
GitHub user lhotari added a comment to the discussion: Is RabbitMQ source connectors safe?
Created issue #24353
GitHub link: https://github.com/apache/pulsar/discussions/23677#discussioncomment-13286621
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
Re: [D] Is RabbitMQ source connectors safe? [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
GitHub user lhotari edited a comment on the discussion: Is RabbitMQ source connectors safe?
@fabrizziocht Thanks for pointing out the issue. The Record interface has [an ack method](https://github.com/apache/pulsar/blob/54f0ca1c769638000fbeb4d73a565216c0992c32/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java#L106-L110) which is where the acknowledgement to the source (RabbitMQ in this case) should be made.
Would you like to create an issue report at https://github.com/apache/pulsar/issues to track addressing this problem?
GitHub link: https://github.com/apache/pulsar/discussions/23677#discussioncomment-11458232
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
Re: [D] Is RabbitMQ source connectors safe? [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
GitHub user lhotari edited a comment on the discussion: Is RabbitMQ source connectors safe?
Thanks for pointing out the issue. The Record interface has [an ack method](https://github.com/apache/pulsar/blob/54f0ca1c769638000fbeb4d73a565216c0992c32/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java#L106-L110) which is where the acknowledgement to the source (RabbitMQ in this case) should be made.
Would you like to create an issue report at https://github.com/apache/pulsar/issues to track addressing this problem?
GitHub link: https://github.com/apache/pulsar/discussions/23677#discussioncomment-11458232
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
Re: [D] Is RabbitMQ source connectors safe? [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
GitHub user lhotari added a comment to the discussion: Is RabbitMQ source connectors safe?
PR is https://github.com/apache/pulsar/pull/24354
GitHub link: https://github.com/apache/pulsar/discussions/23677#discussioncomment-13288879
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
Re: [D] Is RabbitMQ source connectors safe? [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
GitHub user lhotari added a comment to the discussion: Is RabbitMQ source connectors safe?
Thanks for pointing out the issue. The Record interface has [an ack method](https://github.com/apache/pulsar/blob/54f0ca1c769638000fbeb4d73a565216c0992c32/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java#L106-L110) which is where the acknowledgement to the source (RabbitMQ in this case) should be made.
GitHub link: https://github.com/apache/pulsar/discussions/23677#discussioncomment-11458232
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org
Re: [D] Is RabbitMQ source connectors safe? [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
GitHub user lhotari added a comment to the discussion: Is RabbitMQ source connectors safe?
@fabrizziocht do you have a chance to review?
GitHub link: https://github.com/apache/pulsar/discussions/23677#discussioncomment-13288884
----
This is an automatically sent email for commits@pulsar.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@pulsar.apache.org