[
https://issues.apache.org/jira/browse/KAFKA-21313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Charles-Eddy updated KAFKA-21313:
---------------------------------
Description:
Hello, I found this while prototyping key-based ordering for share groups. It
reproduces on plain trunk.
A released batch stays AVAILABLE with an ongoing state transition, and keeps
its acquisition lock, until the persister write completes.
So if the log start offset moves inside that batch in the meantime (retention
or DeleteRecords removing records the group has not consumed yet, seen through
an OFFSET_OUT_OF_RANGE fetch), archiveRecords splits it into per-offset states.
InFlightBatch.maybeInitializeOffsetStateUpdate() copies the batch lock to every
offset, including the AVAILABLE ones, and the ongoing transition is not carried
over, so the write completion never clears these locks.
The next acquire of one of these offsets moves it to ACQUIRED and then throws:
{code:java}
java.lang.IllegalArgumentException: Existing acquisition lock timeout exists,
cannot override.
at
org.apache.kafka.server.share.fetch.InFlightState.updateAcquisitionLockTimeoutTask(InFlightState.java:107)
at
kafka.server.share.SharePartition.acquireSubsetBatchRecords(SharePartition.java:2047)
at kafka.server.share.SharePartition.acquire(SharePartition.java:950){code}
The ShareFetch request fails with UNKNOWN_SERVER_ERROR, and the record stays
ACQUIRED for a member that never received it until the stale lock expires. The
timer task scheduled before the throw is left behind: in my runs it later
released the record while another member was processing it.
The attached patch adds a test that reproduces
it:SharePartitionTest#testReleasedRecordsSplitByLogStartOffsetMoveWhileReleaseWriteIsPendingCanBeAcquired.
Not splitting a batch that has an ongoing transition (or carrying the
transition over), and not copying the lock to offsets that are not ACQUIRED,
should fix it.
was:
Hello, I found this while prototyping key-based ordering for share groups. It
reproduces on plain trunk.
A released batch stays AVAILABLE with an ongoing state transition, and keeps
its acquisition lock, until the persister write completes.
So if the log start offset moves inside that batch in the meantime (retention
or DeleteRecords, seen through an OFFSET_OUT_OF_RANGE fetch), archiveRecords
splits it into per-offset states.
InFlightBatch.maybeInitializeOffsetStateUpdate() copies the batch lock to every
offset, including the AVAILABLE ones, and the ongoing transition is not carried
over, so the write completion never clears these locks.
The next acquire of one of these offsets moves it to ACQUIRED and then throws:
{code:java}
java.lang.IllegalArgumentException: Existing acquisition lock timeout exists,
cannot override.
at
org.apache.kafka.server.share.fetch.InFlightState.updateAcquisitionLockTimeoutTask(InFlightState.java:107)
at
kafka.server.share.SharePartition.acquireSubsetBatchRecords(SharePartition.java:2047)
at kafka.server.share.SharePartition.acquire(SharePartition.java:950){code}
The ShareFetch request fails with UNKNOWN_SERVER_ERROR, and the record stays
ACQUIRED for a member that never received it until the stale lock expires. The
timer task scheduled before the throw is left behind: in my runs it later
released the record while another member was processing it.
The attached patch adds a test that reproduces
it:SharePartitionTest#testReleasedRecordsSplitByLogStartOffsetMoveWhileReleaseWriteIsPendingCanBeAcquired.
Not splitting a batch that has an ongoing transition (or carrying the
transition over), and not copying the lock to offsets that are not ACQUIRED,
should fix it.
> Share partition acquire fails with "Existing acquisition lock timeout exists"
> after a log start offset move splits a batch with a pending release
> -------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-21313
> URL: https://issues.apache.org/jira/browse/KAFKA-21313
> Project: Kafka
> Issue Type: Bug
> Components: core
> Affects Versions: 4.5.0
> Reporter: Charles-Eddy
> Assignee: Apoorv Mittal
> Priority: Minor
> Attachments: 1-stale-acquisition-lock-after-split.patch
>
>
> Hello, I found this while prototyping key-based ordering for share groups. It
> reproduces on plain trunk.
> A released batch stays AVAILABLE with an ongoing state transition, and keeps
> its acquisition lock, until the persister write completes.
> So if the log start offset moves inside that batch in the meantime (retention
> or DeleteRecords removing records the group has not consumed yet, seen
> through an OFFSET_OUT_OF_RANGE fetch), archiveRecords splits it into
> per-offset states.
> InFlightBatch.maybeInitializeOffsetStateUpdate() copies the batch lock to
> every offset, including the AVAILABLE ones, and the ongoing transition is not
> carried over, so the write completion never clears these locks.
> The next acquire of one of these offsets moves it to ACQUIRED and then throws:
> {code:java}
> java.lang.IllegalArgumentException: Existing acquisition lock timeout exists,
> cannot override.
> at
> org.apache.kafka.server.share.fetch.InFlightState.updateAcquisitionLockTimeoutTask(InFlightState.java:107)
> at
> kafka.server.share.SharePartition.acquireSubsetBatchRecords(SharePartition.java:2047)
> at
> kafka.server.share.SharePartition.acquire(SharePartition.java:950){code}
> The ShareFetch request fails with UNKNOWN_SERVER_ERROR, and the record stays
> ACQUIRED for a member that never received it until the stale lock expires.
> The timer task scheduled before the throw is left behind: in my runs it later
> released the record while another member was processing it.
> The attached patch adds a test that reproduces
> it:SharePartitionTest#testReleasedRecordsSplitByLogStartOffsetMoveWhileReleaseWriteIsPendingCanBeAcquired.
> Not splitting a batch that has an ongoing transition (or carrying the
> transition over), and not copying the lock to offsets that are not ACQUIRED,
> should fix it.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
