KAFKA-15802: validate remote segment state before fetching index #14727

jeqo · 2023-11-09T20:59:37Z

See commits for more details

[KAFKA-15802]

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

showuon

LGTM!

kamalcph

LGTM

divijvaidya

You will need a similar filter in cleanupExpiredRemoteLogSegments() method as well where call listRemoteLogSegments() to fetch the metadata. Over there we want to iterate through segments which are in copy_finished state OR delete_started state (otherwise same error you faced in this JIRA will surface there).
Please add integration test which fails prior to this change for both the changes, one which you have added and another which I suggested in point 1 above.

jeqo · 2023-11-13T14:35:01Z

@divijvaidya good catch!
Although, in the case of delete there's no exception as deleting a segment that is not present on remote does not throw an exception as it's idempotent. Regardless, it make sense to add this validation to save us a remote call and only call delete segment when appropriate.

divijvaidya · 2023-11-13T15:46:46Z

core/src/main/java/kafka/log/remote/RemoteLogManager.java

-                    // since the checkpoint file was already truncated.
-                    boolean shouldDeleteSegment = remoteLogRetentionHandler.isSegmentBreachByLogStartOffset(
+
+                    if (SEGMENT_DELETION_VALID_STATES.contains(metadata.state())) {


since there is no "else" part to it, perhaps short circuiting the if condition with

if (!SEGMENT_DELETION_VALID_STATES.contains(metadata.state())) { continue; }

would provide more readability.

(I don't have a strong opinion on this, hence, let me know if you wish to keep it as as it)

Agree, applying this fix.

divijvaidya · 2023-11-13T16:02:10Z

core/src/main/java/kafka/log/remote/RemoteLogManager.java

@@ -133,6 +134,10 @@ public class RemoteLogManager implements Closeable {

    private static final Logger LOGGER = LoggerFactory.getLogger(RemoteLogManager.class);
    private static final String REMOTE_LOG_READER_THREAD_NAME_PREFIX = "remote-log-reader";
+    private static final Set<RemoteLogSegmentState> SEGMENT_DELETION_VALID_STATES = Collections.unmodifiableSet(EnumSet.of(


In this diagram:

kafka/storage/api/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManager.java

Line 74 in 49d3122

* +---------------------+ +----------------------+

copy_started -> delete_started is also a valid transition. Although, I can't think of a scenario where this would be valid because we always complete copy before calling expiration. Are we missing something here by not including copy_started?

Great catch. I guess the transition hasn't considered the current implementation where copying and deletion are sequential. If the implementation changes, it may be possible to have this scenario.
I'd say it should be considered a valid transition. I will include it in the valid states and adapt the test case.

COPY_SEGMENT_STARTED segments are eligible for deletion when those segments were not able to be copied by the leader as the leader went through ungraceful shutdown or for any oher reasons. New leader may pickup the resepctive segments for the targeted offsets that need to be copied and the earlier failed segment will remain in the COPY_SEGMENT_STARTED state and it will eventually be deleted by retention cleanup logic.

So, COPY_SEGMENT_STARTED is a valid transition even now when copy and deletion are happening sequentially.

@satishd in that case wouldn't the new leader retry to upload the segment? If a segment is left on copy_started state, then it'd risk some segment not being uploaded. Is this expected?

@jeqo Leader will upload the required segment with a new UUID by finding the offset that needs to be copied as mentioned here. It will not change the state of the failed copy of an earlier segment.

Aha! I can see a new segment id being generated here.
Got it, so there may be segment that will stay on copy_started until eventually deleted. Then, the way to assess if this is not an issue is to check that another topic-id-partition-offset combination with a different segment id has copy_finished, isn't it?

Right, one way to check is whether there is a segment associated with the target offset or not by using RLMM#remoteLogSegmentMetadata().

satishd

Thanks @jeqo for the PR, Left a couple of comment/clarifications.

satishd · 2023-11-14T06:12:02Z

core/src/main/java/kafka/log/remote/RemoteLogManager.java

@@ -133,6 +134,10 @@ public class RemoteLogManager implements Closeable {

    private static final Logger LOGGER = LoggerFactory.getLogger(RemoteLogManager.class);
    private static final String REMOTE_LOG_READER_THREAD_NAME_PREFIX = "remote-log-reader";
+    private static final Set<RemoteLogSegmentState> SEGMENT_DELETION_VALID_STATES = Collections.unmodifiableSet(EnumSet.of(


COPY_SEGMENT_STARTED segments are eligible for deletion when those segments were not able to be copied by the leader as the leader went through ungraceful shutdown or for any oher reasons. New leader may pickup the resepctive segments for the targeted offsets that need to be copied and the earlier failed segment will remain in the COPY_SEGMENT_STARTED state and it will eventually be deleted by retention cleanup logic.

So, COPY_SEGMENT_STARTED is a valid transition even now when copy and deletion are happening sequentially.

core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java

Currently, the RLMM is fetching the index even when remote segment metadata state says upload has not complete yet. Workaround using current APIs to only fetch remote indexes when segment is properly uploaded. Also, the RLMM is deleting segments even when remote segment metadata state says upload has not complete yet. Only consider segments where copy has started or completed, or where delete has started. [KAFKA-15802](https://issues.apache.org/jira/browse/KAFKA-15802)

satishd · 2023-11-14T08:34:18Z

@jeqo Please maintain the commit history if possible as it helps in reviewing the updated commits in the PR. It is fine if you need to rebase and merge the commits for valid reasons.

kamalcph

Overall LGTM, left minor comments.

kamalcph · 2023-11-14T09:02:07Z

core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java

        verifyDeleteLogSegment(segmentMetadataList, deletableSegmentCount, currentLeaderEpoch);
    }

    @ParameterizedTest(name = "testDeleteLogSegmentDueToRetentionTimeBreach segmentCount={0} deletableSegmentCount={1}")
-    @CsvSource(value = {"50, 0", "50, 1", "50, 23", "50, 50"})
+    @CsvSource(value = {"50, 0, DELETE_SEGMENT_STARTED", "50, 1, COPY_SEGMENT_FINISHED", "50, 23, DELETE_SEGMENT_STARTED", "50, 50, COPY_SEGMENT_FINISHED"})


can we skip updating the existing tests?

There is already a test (testDeleteRetentionMsBeforeSegmentReady) to verify that if the state is DELETE_SEGMENT_FINISHED, no interaction is expected with the RemoteStorageManager.

Can we also rename the test?

testDeleteRetentionMsBeforeSegmentReady to testDeleteRetentionMsOnExpiredSegment (expired/invalid/something-similar)

Agree, applying suggestion

kamalcph · 2023-11-14T09:06:38Z

core/src/main/java/kafka/log/remote/RemoteLogManager.java

@@ -988,6 +998,10 @@ void cleanupExpiredRemoteLogSegments() throws RemoteStorageException, ExecutionE
                        return;
                    }
                    RemoteLogSegmentMetadata metadata = segmentsIterator.next();
+
+                    if (!SEGMENT_DELETION_VALID_STATES.contains(metadata.state())) {


Can this condition be simplified to?

if (RemoteLogSegmentState.DELETE_SEGMENT_FINISHED.equals(metadata.state())) { continue; }

If we want to add/remove state, we can change the logic later.

Good idea, applying change.

…ex/cleanup Align code style

…ex/cleanup Apply latest suggestions

kamalcph

LGTM!

satishd

Thanks @jeqo for addressing the review comments. LGTM.

satishd · 2023-11-14T16:24:15Z

There are a few unrelated test failures, merging the changes to 3.6 branch.

@jeqo Please raise PR against trunk by picking this change and let me know, will review and merge it when there are no related failures in Jenkins job.

…che#14727) Reviewers: Satish Duggana <satishd@apache.org>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Luke Chen <showuon@gmail.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

) (#14759) Reviewers: Satish Duggana <satishd@apache.org>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Luke Chen <showuon@gmail.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

…che#14727) Reviewers: Satish Duggana <satishd@apache.org>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Luke Chen <showuon@gmail.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

…che#14727) (apache#14759) Reviewers: Satish Duggana <satishd@apache.org>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Luke Chen <showuon@gmail.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

…che#14727) (apache#14759) (#244) Reviewers: Satish Duggana <satishd@apache.org>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Luke Chen <showuon@gmail.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com> Co-authored-by: Jorge Esteban Quilcate Otoya <jorge.quilcate@aiven.io>

…che#14727) (apache#14759) Reviewers: Satish Duggana <satishd@apache.org>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Luke Chen <showuon@gmail.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

showuon approved these changes Nov 10, 2023

View reviewed changes

satishd self-requested a review November 10, 2023 08:09

kamalcph approved these changes Nov 11, 2023

View reviewed changes

hudeqi self-requested a review November 13, 2023 02:46

divijvaidya added the tiered-storage Pull requests associated with KIP-405 (Tiered Storage) label Nov 13, 2023

divijvaidya reviewed Nov 13, 2023

View reviewed changes

divijvaidya added the backport-candidate This pull request is a candidate to be backported to previous versions label Nov 13, 2023

jeqo force-pushed the jeqo/kafka-15802 branch 2 times, most recently from ec765c7 to cd6b1f8 Compare November 13, 2023 14:34

jeqo requested review from showuon, kamalcph and divijvaidya November 13, 2023 14:35

divijvaidya reviewed Nov 13, 2023

View reviewed changes

jeqo force-pushed the jeqo/kafka-15802 branch from cd6b1f8 to 864cdd1 Compare November 13, 2023 23:09

satishd reviewed Nov 14, 2023

View reviewed changes

jeqo force-pushed the jeqo/kafka-15802 branch from 864cdd1 to 6140a9d Compare November 14, 2023 07:55

kamalcph approved these changes Nov 14, 2023

View reviewed changes

jeqo added 2 commits November 14, 2023 11:47

fixup! KAFKA-15802: validate remote segment state before fetching ind…

da3c854

…ex/cleanup Align code style

fixup! KAFKA-15802: validate remote segment state before fetching ind…

18a7c4f

…ex/cleanup Apply latest suggestions

jeqo requested a review from kamalcph November 14, 2023 09:49

kamalcph approved these changes Nov 14, 2023

View reviewed changes

divijvaidya approved these changes Nov 14, 2023

View reviewed changes

satishd approved these changes Nov 14, 2023

View reviewed changes

satishd merged commit 5fa2a24 into apache:3.6 Nov 14, 2023
1 check failed

jeqo deleted the jeqo/kafka-15802 branch November 14, 2023 17:18

jeqo restored the jeqo/kafka-15802 branch November 14, 2023 17:19

jeqo mentioned this pull request Nov 14, 2023

KAFKA-15802: Validate remote segment state before fetching index #14759

Merged

3 tasks

jeqo deleted the jeqo/kafka-15802 branch November 14, 2023 19:24

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-15802: validate remote segment state before fetching index #14727

KAFKA-15802: validate remote segment state before fetching index #14727

jeqo commented Nov 9, 2023

showuon left a comment

kamalcph left a comment

divijvaidya left a comment •

edited

jeqo commented Nov 13, 2023

divijvaidya Nov 13, 2023

jeqo Nov 13, 2023

divijvaidya Nov 13, 2023

jeqo Nov 13, 2023

satishd Nov 14, 2023

jeqo Nov 14, 2023

satishd Nov 14, 2023

jeqo Nov 14, 2023

satishd Nov 14, 2023

satishd left a comment

satishd Nov 14, 2023

satishd commented Nov 14, 2023

kamalcph left a comment

kamalcph Nov 14, 2023

kamalcph Nov 14, 2023

jeqo Nov 14, 2023

kamalcph Nov 14, 2023

jeqo Nov 14, 2023

kamalcph left a comment

satishd left a comment

satishd commented Nov 14, 2023 •

edited

KAFKA-15802: validate remote segment state before fetching index #14727

KAFKA-15802: validate remote segment state before fetching index #14727

Conversation

jeqo commented Nov 9, 2023

Committer Checklist (excluded from commit message)

showuon left a comment

Choose a reason for hiding this comment

kamalcph left a comment

Choose a reason for hiding this comment

divijvaidya left a comment • edited

Choose a reason for hiding this comment

jeqo commented Nov 13, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

satishd commented Nov 14, 2023

kamalcph left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kamalcph left a comment

Choose a reason for hiding this comment

satishd left a comment

Choose a reason for hiding this comment

satishd commented Nov 14, 2023 • edited

divijvaidya left a comment •

edited

satishd commented Nov 14, 2023 •

edited