Follow-up from the review of #9076.
Since #9076, a worker drops a message that deterministically fails to decode instead of dying. The dropped tuple is never acked by its receiver, so with the kafka spout in at-least-once mode its offset never commits. Polling continues past the undecodable record, every later offset stays uncommittable behind it, uncommitted offsets climb to maxUncommittedOffsets (default 10,000,000), and getPollablePartitionsInfo then stops polling the partition: a quiet per-partition stall whose symptom is climbing lag. Before #9076 the same input was a worker crash loop, which also never committed the offset but was loud.
The building blocks for an eventual permanent drop already exist. With a finite maxRetries, once the retry service gives up, KafkaSpout.fail marks the message as acked (the branch that calls tupleListener.onMaxRetryReached) and the OffsetManager lets the commit progress again. But the default is to retry forever, and when the spout gives up it acks blind: it holds the raw ConsumerRecord and could dead-letter it, yet today the record just disappears.
What I think deserves design attention: whether the receiver can tell the spout that a message was dropped as undecodable, since today the failure only shows up as the deserializationFailures metric and worker logs while the spout side just sees a tuple that never gets acked; and whether giving up permanently should dead-letter the raw record rather than ack it silently. Default guidance for maxRetries in the presence of persistent decode failures may be all that is needed.
Follow-up from the review of #9076.
Since #9076, a worker drops a message that deterministically fails to decode instead of dying. The dropped tuple is never acked by its receiver, so with the kafka spout in at-least-once mode its offset never commits. Polling continues past the undecodable record, every later offset stays uncommittable behind it, uncommitted offsets climb to maxUncommittedOffsets (default 10,000,000), and getPollablePartitionsInfo then stops polling the partition: a quiet per-partition stall whose symptom is climbing lag. Before #9076 the same input was a worker crash loop, which also never committed the offset but was loud.
The building blocks for an eventual permanent drop already exist. With a finite maxRetries, once the retry service gives up, KafkaSpout.fail marks the message as acked (the branch that calls tupleListener.onMaxRetryReached) and the OffsetManager lets the commit progress again. But the default is to retry forever, and when the spout gives up it acks blind: it holds the raw ConsumerRecord and could dead-letter it, yet today the record just disappears.
What I think deserves design attention: whether the receiver can tell the spout that a message was dropped as undecodable, since today the failure only shows up as the deserializationFailures metric and worker logs while the spout side just sees a tuple that never gets acked; and whether giving up permanently should dead-letter the raw record rather than ack it silently. Default guidance for maxRetries in the presence of persistent decode failures may be all that is needed.