KAFKA-14993: Improve TransactionIndex instance handling while copying to and fetching from RSM #14363

abhijeetk88 · 2023-09-09T14:09:22Z

In this PR, we have updated the contract for RSM's fetchIndex to throw a ResourceNotFoundException if it does not have TransactionIndex, instead of returning an empty inputstream.

Also, updated the LocalTieredStorage implementation to adhere to the new contract.

Added Unit Tests for the change.

Committer Checklist (excluded from commit message)

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

… to and fetching from RSM.

satishd

Thanks @abhijeetk88 for the PR. Left a comment for clarification.

core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala

satishd

Thanks @abhijeetk88 for addressing the review comments.

satishd · 2023-09-10T03:51:06Z

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

@@ -123,7 +123,7 @@ InputStream fetchLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
     * Returns the index for the respective log segment of {@link RemoteLogSegmentMetadata}.
     * <p>
     * Note: The transaction index may not exist because of no transactional records.


I am fine to leave this comment for better clarity even though param doc clarifies that RemoteResourceNotFoundException is thrown when any index is not found in remote storage.

Good to me, too.

satishd · 2023-09-10T04:08:28Z

cc @ivanyu , as you requested it for 3.6.0

satishd

Thanks @abhijeetk88 for addressing the comments. LGTM.

showuon · 2023-09-11T03:23:44Z

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

@@ -123,7 +123,7 @@ InputStream fetchLogSegment(RemoteLogSegmentMetadata remoteLogSegmentMetadata,
     * Returns the index for the respective log segment of {@link RemoteLogSegmentMetadata}.
     * <p>
     * Note: The transaction index may not exist because of no transactional records.


Good to me, too.

showuon · 2023-09-11T03:27:18Z

storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java

@@ -382,6 +385,8 @@ private RemoteIndexCache.Entry createCacheEntry(RemoteLogSegmentMetadata remoteL
            TransactionIndex txnIndex = loadIndexFile(txnIndexFile, remoteLogSegmentMetadata, rlsMetadata -> {
                try {
                    return remoteStorageManager.fetchIndex(rlsMetadata, IndexType.TRANSACTION);
+                } catch (RemoteResourceNotFoundException e) {
+                    return null;


We'd better to add some comment here to explain why we treat txnIndex differently. Ex:
Don't throw exception since the transaction index may not exist because of no transactional records.

showuon · 2023-09-11T03:30:24Z

storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java

@@ -451,7 +456,7 @@ public static class Entry implements AutoCloseable {

        private final OffsetIndex offsetIndex;
        private final TimeIndex timeIndex;
-        private final TransactionIndex txnIndex;
+        private final Optional<TransactionIndex> txnIndexOpt;


We should make it clear again why we treat txnIndex differently. Please add a comment. Thanks.

ivanyu · 2023-09-11T04:33:31Z

Thank you @abhijeetk88!
Apart from what's requested by @showuon, LGTM

abhijeetk88 · 2023-09-11T05:14:19Z

@showuon Thanks for review. I have addressed your comments. Please take a look.

showuon

LGTM!

kamalcph

LGTM

kamalcph

Overall LGTM, left minor comment to address. Thanks!

divijvaidya

During the initial load in init() method, we have a condition where we load existing entries on disk into the cache based on the condition:

if (Files.exists(offsetIndexFile.toPath()) &&
                            Files.exists(timestampIndexFile.toPath()) &&
                            Files.exists(txnIndexFile.toPath()))

Since, trxIndex is optional, for most cases, this will end up deleting these pre-existing index files and loading them again from remote. This is the safe way because we don't know if the trxIndex is genuinely missing because it doesn't exist for this segment or it's missing due to some error.

May I suggest that we use an empty file on disk to specify that trxIndex is genuinely absent (and not because of some error)? This would help make the interface more deterministic since now we can differentiate cases genuine vs. error scenario.

@showuon @satishd thoughts?

storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java

divijvaidya · 2023-09-11T10:25:41Z

storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java

@@ -382,6 +388,9 @@ private RemoteIndexCache.Entry createCacheEntry(RemoteLogSegmentMetadata remoteL
            TransactionIndex txnIndex = loadIndexFile(txnIndexFile, remoteLogSegmentMetadata, rlsMetadata -> {
                try {
                    return remoteStorageManager.fetchIndex(rlsMetadata, IndexType.TRANSACTION);
+                } catch (RemoteResourceNotFoundException e) {
+                    // Don't throw an exception since the transaction index may not exist because of no transactional records.
+                    return null;


please add a javadoc for the usage of this method that null is expected

Do you mean we should add javadoc to the below method? This one is the caller and the second parameter could receive null input stream.

private <T> T loadIndexFile(File file, RemoteLogSegmentMetadata remoteLogSegmentMetadata, Function<RemoteLogSegmentMetadata, InputStream> fetchRemoteIndex, Function<File, T> readIndex) throws IOException {

The method is private. Should we still add a javadoc?

yes please. I know it is not necessary for private methods but it would be useful for future code implementors to understand that they are expected to handle nulls returned by this method.

I got confused with this statement:

This one is the caller and the second parameter could receive null input stream.

It's the third parameter. Can you place the one parameter per line for readability?

satishd · 2023-09-11T11:44:09Z

@divijvaidya I do not have a strong opinion on that, I am fine with either way. Sure, we can create an empty file in the file cache if there is no txnIndex for a specific segment.

showuon · 2023-09-11T12:50:50Z

Since, trxIndex is optional, for most cases, this will end up deleting these pre-existing index files and loading them again from remote. This is the safe way because we don't know if the trxIndex is genuinely missing because it doesn't exist for this segment or it's missing due to some error.

May I suggest that we use an empty file on disk to specify that trxIndex is genuinely absent (and not because of some error)? This would help make the interface more deterministic since now we can differentiate cases genuine vs. error scenario.

Nice catch! Yes, this is indeed an issue. For your suggestion @divijvaidya , I was thinking about how we could distinguish between "real empty txnInex file" and the one with absent txnIndex file. Then, I realize we don't have to care about that because in both cases, we basically will do no-op when getting the file. So, I'm +1 to store an empty txnIndex file instead of null in the cache entry.

satishd

Thanks @abhijeetk88 for addressing the latest review comments in the updated PR, that includes the test changes for checking empty files. LGTM.

abhijeetk88 · 2023-09-12T06:33:16Z

@divijvaidya @showuon I have addressed your comments. Please take a look.

kamalcph

Changes looks clean with the latest commit. Thanks, @abhijeetk88 for working on this! LGTM.

satishd · 2023-09-12T12:20:49Z

There are unrelated test failures, merging it to trunk and 3.6.

… to and fetching from RSM (#14363) - Updated the contract for RSM's fetchIndex to throw a ResourceNotFoundException instead of returning an empty InputStream when it does not have a TransactionIndex. - Updated the LocalTieredStorage implementation to adhere to the new contract. - Added Unit Tests for the change. Reviewers: Satish Duggana <satishd@apache.org>, Luke Chen <showuon@gmail.com>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

… to and fetching from RSM (apache#14363) - Updated the contract for RSM's fetchIndex to throw a ResourceNotFoundException instead of returning an empty InputStream when it does not have a TransactionIndex. - Updated the LocalTieredStorage implementation to adhere to the new contract. - Added Unit Tests for the change. Reviewers: Satish Duggana <satishd@apache.org>, Luke Chen <showuon@gmail.com>, Divij Vaidya <diviv@amazon.com>, Christo Lolov <lolovc@amazon.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

abhijeetk88 marked this pull request as draft September 9, 2023 14:09

satishd added the tiered-storage Pull requests associated with KIP-405 (Tiered Storage) label Sep 9, 2023

KAFKA-14993: Improve TransactionIndex instance handling while copying…

07378c3

… to and fetching from RSM.

abhijeetk88 force-pushed the KAFKA-14993 branch from 22b82d3 to 07378c3 Compare September 9, 2023 18:41

abhijeetk88 marked this pull request as ready for review September 9, 2023 18:42

satishd reviewed Sep 10, 2023

View reviewed changes

core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala Show resolved Hide resolved

satishd reviewed Sep 10, 2023

View reviewed changes

apache deleted a comment from Lalolokos1 Sep 10, 2023

satishd requested a review from showuon September 10, 2023 09:38

satishd approved these changes Sep 11, 2023

View reviewed changes

showuon reviewed Sep 11, 2023

View reviewed changes

Address review comments

14b949e

Fix spotbugs error

b7c8101

showuon approved these changes Sep 11, 2023

View reviewed changes

kamalcph approved these changes Sep 11, 2023

View reviewed changes

kamalcph suggested changes Sep 11, 2023

View reviewed changes

kamalcph approved these changes Sep 11, 2023

View reviewed changes

divijvaidya reviewed Sep 11, 2023

View reviewed changes

Create an empty file if there is not transactional records in remote

778d05a

abhijeetk88 force-pushed the KAFKA-14993 branch from d01fc72 to 778d05a Compare September 12, 2023 06:00

satishd approved these changes Sep 12, 2023

View reviewed changes

kamalcph approved these changes Sep 12, 2023

View reviewed changes

divijvaidya approved these changes Sep 12, 2023

View reviewed changes

showuon approved these changes Sep 12, 2023

View reviewed changes

clolov approved these changes Sep 12, 2023

View reviewed changes

satishd merged commit a63bf93 into apache:trunk Sep 12, 2023
1 check failed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-14993: Improve TransactionIndex instance handling while copying to and fetching from RSM #14363

KAFKA-14993: Improve TransactionIndex instance handling while copying to and fetching from RSM #14363

abhijeetk88 commented Sep 9, 2023 •

edited

satishd left a comment

satishd left a comment

satishd Sep 10, 2023

showuon Sep 11, 2023

satishd commented Sep 10, 2023 •

edited

satishd left a comment

showuon Sep 11, 2023

showuon Sep 11, 2023

abhijeetk88 Sep 11, 2023

showuon Sep 11, 2023

abhijeetk88 Sep 11, 2023

ivanyu commented Sep 11, 2023

abhijeetk88 commented Sep 11, 2023

showuon left a comment

kamalcph left a comment

kamalcph left a comment

divijvaidya left a comment

divijvaidya Sep 11, 2023

abhijeetk88 Sep 11, 2023

divijvaidya Sep 11, 2023

kamalcph Sep 12, 2023

satishd commented Sep 11, 2023

showuon commented Sep 11, 2023

satishd left a comment

abhijeetk88 commented Sep 12, 2023

kamalcph left a comment

satishd commented Sep 12, 2023

KAFKA-14993: Improve TransactionIndex instance handling while copying to and fetching from RSM #14363

KAFKA-14993: Improve TransactionIndex instance handling while copying to and fetching from RSM #14363

Conversation

abhijeetk88 commented Sep 9, 2023 • edited

Committer Checklist (excluded from commit message)

satishd left a 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

Choose a reason for hiding this comment

satishd commented Sep 10, 2023 • edited

satishd 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

ivanyu commented Sep 11, 2023

abhijeetk88 commented Sep 11, 2023

showuon left a comment

Choose a reason for hiding this comment

kamalcph 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

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 commented Sep 11, 2023

showuon commented Sep 11, 2023

satishd left a comment

Choose a reason for hiding this comment

abhijeetk88 commented Sep 12, 2023

kamalcph left a comment

Choose a reason for hiding this comment

satishd commented Sep 12, 2023

abhijeetk88 commented Sep 9, 2023 •

edited

satishd commented Sep 10, 2023 •

edited