You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
awaitingCommitResult should be incremented and subtracted by the same value to become zero eventually and allow clean shutdown.
In produceawaitingCommitResult should be incremented by msg.passThrough.batchSize and multiMsg.passThrough.batchSize correspondingly.
Actual Behavior
During the shutdown CommittingProducerSinkStageLogic awaits not finished commits by checking awaitingCommitResult is zero.
In produce this value is always incremented by 1 when single message is committed using Committable and when many messages are committed by CommittableOffsetBatch.
In commitResultCB on the other hand batchSize is subtracted. This leads awaitingCommitResult to be always negative.
As a result during the shutdown it cannot terminate correctly.
In my case I'm aggregating incoming messages by 100 using groupedWithin operator. Next from all grouped messages I'm building one CommittableOffsetBatch. CommittingProducerSinkStageLogic increments awaitingCommitResult by 1 for every batch. All batches are aggregated in even bigger batch. When such batch of size 400-500 is committed awaitingCommitResult is subtracted by the size of the batch.
The text was updated successfully, but these errors were encountered:
I agree, it looks like as if I mixed the counters for awaitingProduceResult and awaitingCommitResult in the callback. This should be reproducible in a test case.
Versions used
Akka version: 2.6.19
Alpakka Kafka version: 3.0.1
Expected Behavior
awaitingCommitResult
should be incremented and subtracted by the same value to become zero eventually and allow clean shutdown.In
produce
awaitingCommitResult
should be incremented bymsg.passThrough.batchSize
andmultiMsg.passThrough.batchSize
correspondingly.Actual Behavior
During the shutdown
CommittingProducerSinkStageLogic
awaits not finished commits by checkingawaitingCommitResult
is zero.In
produce
this value is always incremented by 1 when single message is committed usingCommittable
and when many messages are committed byCommittableOffsetBatch
.In
commitResultCB
on the other handbatchSize
is subtracted. This leadsawaitingCommitResult
to be always negative.As a result during the shutdown it cannot terminate correctly.
In my case I'm aggregating incoming messages by 100 using
groupedWithin
operator. Next from all grouped messages I'm building oneCommittableOffsetBatch
.CommittingProducerSinkStageLogic
incrementsawaitingCommitResult
by 1 for every batch. All batches are aggregated in even bigger batch. When such batch of size 400-500 is committedawaitingCommitResult
is subtracted by the size of the batch.The text was updated successfully, but these errors were encountered: