[
https://issues.apache.org/jira/browse/FLINK-2624?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14738549#comment-14738549
]
Robert Metzger commented on FLINK-2624:
---------------------------------------
Why can't you just extend from {{RichSourceFunction}} ? (and maybe implement
{{implements ResultTypeQueryable<OUT>}} if needed?)
In the {{run()}} method, you need to synchronize the state update
(lastProcessedMessage = delivery.getEnvelope().getDeliveryTag()) and passing
the element to the output collector, using the checkpoint lock from the source
context. Otherwise, it can happen that the processing is not exactly once (we
need to lock the while(running) loop while calling snapshotState()).
> RabbitMQ source / sink should participate in checkpointing
> ----------------------------------------------------------
>
> Key: FLINK-2624
> URL: https://issues.apache.org/jira/browse/FLINK-2624
> Project: Flink
> Issue Type: Bug
> Components: Streaming Connectors
> Affects Versions: 0.10
> Reporter: Stephan Ewen
> Assignee: Hilmi Yildirim
>
> The RabbitMQ connector does not offer any fault tolerance guarantees right
> now, because it does not participate in the checkpointing.
> We should integrate it in a similar was as the {{FlinkKafkaConsumer}} is
> integrated.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
