[
https://issues.apache.org/jira/browse/KAFKA-7480?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16743252#comment-16743252
]
John Roesler commented on KAFKA-7480:
-------------------------------------
Hi [~mjsax], just to confirm the scope of work here, after reading the other
ticket, it sounds like we just need to set the policy to "none", catch the
InvalidOffsetException, and in that case, seek to the beginning and rebuild the
store accordingly.
But your latter comment confused me. What could be a more resilient
configuration than starting from the beginning and rebuilding the store? From
my knowledge of the InvalidOffsetException, it happens when a state store
doesn't consume its input topic for a long time (eg., the task was assigned to
a different node for a while and then migrates back, so the store hasn't kept
up), and its "current offset" has already been deleted by the retention policy.
I suppose that if we know this has happened, then we could actually keep the
store, but seek to the beginning (verifying that the beginning offset is larger
than our invalid "current" offset) and consume from there. If the topic is
compacted, then anything that needs to be updated will get updated in course,
and anything that needs to stay the same will idempotently remain the same. The
only problem would be anything that needs to be tombstoned, but I think we keep
the tombstones in the compacted topic for a much longer time for this reason.
Was this what you had in mind?
It seems like we have several choices for scope:
A: smaller: just make the global store behave like a local store and rebuild
from the beginning when it gets the exception
B: medium: consider optimizing by only discarding and rebuilding the state for
the partition that gets the exception. Eg., we could keep each partition in a
separate column family (CF) and employ a localized CF drop-and-restore
strategy. This would depend on knowing the partition function so we can route
queries both from Streams operators and from IQ.
C: medium: try to keep the existing store, and still rebuild on top of it. This
*should* be fine, but requires some detail-oriented analysis to make sure we
won't produce corrupted state. This same strategy could apply to local stores
as well, and in fact is orthogonal to B. (We could do both B and C).
> GlobalThread should honor custom auto.offset.reset policy
> ---------------------------------------------------------
>
> Key: KAFKA-7480
> URL: https://issues.apache.org/jira/browse/KAFKA-7480
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: Matthias J. Sax
> Priority: Major
> Labels: needs-kip
>
> With KAFKA-6121 we improved Kafka Streams resilience and correctness with
> regard to consumer auto.offset.reset and state cleanup.
> Back than, we decided to let GlobalStreamThread die and not handle
> InvalidOffsetException during regular processing, because this error
> indicates a fatal issue and the user should be notified about it. However, as
> reported on the user mailing list, the only thing a user can do is, to
> restart the application (and investigate the root cause). During restart, the
> state will be cleaned up and bootstrapped correctly.
> Thus, we might want to allow users to specify a more resilient configuration
> for this case and log an ERROR message if the error occurs. To ensure
> consistency, we might not allow to set reset policy "latest" though (need
> discussion)? By default, we can still keep "none" and fail.
> Note: `Topology.addGlobalStore` does not allow to set a reset policy. Thus,
> this might require a KIP to extend `Topology.addGlobalStore` accordingly.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
