Fix: function with multi-topic not acking on effectively-once #2347
Conversation
rdhabalia
added
type/bug
| if (pulsarSourceConfig | ||
| .getProcessingGuarantees() == FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE) { | ||
| // try to find actual consumer of the messageId | ||
| if (inputConsumer instanceof MultiTopicsConsumerImpl) { |
There was a problem hiding this comment.
does this mean ackCumulativeAsync is not working in MultiTopicsConsumerImpl? If so, should this be fixed in MultiTopicsConsumerImpl instead?
There was a problem hiding this comment.
If so, should this be fixed in MultiTopicsConsumerImpl instead?
Yeah, earlier I was thinking about ordering but MultiConsumer anyway doesn't maintain ordering among the consumers and we can also do cumulative ack for individual consumer. let me fix it.
|
@sijie fixed it. |
|
rerun java8 tests |
|
@rdhabalia merged. I also changed the milestone to 2.2.0 since it is merged to master. when it is cherry-picked to 2.1.1, will update the milestone. otherwise, let's assume the changes based on master will be parted to major releases. |
|
sure.. 👍 |
…#2347) ### Motivation `MultiTopicsConsumerImpl` doesn't support `acknowledgeCumulativeAsync` and therefore, function with multi-topic and `EFFECTIVELY_ONCE` processing is not acking message and failing `EFFECTIVELY_ONCE` behavior. ### Modifications Function should ack message for a specific topic consumer if `inputTopicConsumer` is multi-topic consumer. ### Result Function should able to ack messages for multi-topic consumer when processing-guarantee is `EFFECTIVELY_ONCE`
|
merged 068db59 into branch-2.1 |
Motivation
MultiTopicsConsumerImpldoesn't supportacknowledgeCumulativeAsyncand therefore, function with multi-topic andEFFECTIVELY_ONCEprocessing is not acking message and failingEFFECTIVELY_ONCEbehavior.Modifications
Function should ack message for a specific topic consumer if
inputTopicConsumeris multi-topic consumer.Result
Function should able to ack messages for multi-topic consumer when processing-guarantee is
EFFECTIVELY_ONCE