KAFKA-15169: Added TestCase in RemoteIndexCache #14482

iit2009060 · 2023-10-04T04:45:32Z

Test Cases Covered

Index Files already exist on disk but not in Cache i.e. RemoteIndexCache should not call remoteStorageManager to fetch it instead cache it from the local index file present.
RSM returns CorruptedIndex File i.e. RemoteIndexCache should throw CorruptedIndexException instead of successfull execution.
Deleted Suffix Indexes file already present on disk i.e. If cleaner thread is slow , then there is a chance of deleted index files present on the disk while in parallel same index Entry is invalidated. To understand more refer https://issues.apache.org/jira/browse/KAFKA-15169

Committer Checklist (excluded from commit message)

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

divijvaidya · 2023-10-04T07:54:55Z

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

+  def testOffsetIndexFileAlreadyExistOnDiskButNotInCache(): Unit ={
+    val remoteIndexCacheDir = new File(tpDir,RemoteIndexCache.DIR_NAME)


Please follow style guide used in rest of the code. For example, space between = and {.
You can find more information here: https://kafka.apache.org/coding-guide.html

I remember your previous PR also had similar comments. I would recommend to set your IDE to use this style and automatically format newly added code.

Thanks , I have enabled on IDE to format it on save everytime.

divijvaidya · 2023-10-04T07:57:40Z

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

+  def testOffsetIndexFileAlreadyExistOnDiskButNotInCache(): Unit ={
+    val remoteIndexCacheDir = new File(tpDir,RemoteIndexCache.DIR_NAME)


please create a getter in the RemoteIndexCache to get this dir, add javadoc for it // visible for testing and use that here. It will make this test independent of the remote cache location in case we change that in future,

divijvaidya · 2023-10-04T07:58:44Z

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

+      Files.walk(remoteIndexCacheDir.toPath)
+        .filter(Files.isRegularFile(_))
+        .filter(path => path.getFileName.toString.endsWith(suffix))
+        .forEach(f => Files.move(f,f.resolveSibling(f.getFileName().toString().stripSuffix(tempSuffix))))


Instead of files.move, please use our custom utility function Utils.atomicMoveWithFallback

divijvaidya · 2023-10-04T08:05:46Z

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

+    // rsm should not be called again
+    // instead files exist on disk
+    // should be used
+    verifyFetchIndexInvocation(count = 1)


should this be 0 if there is going to be no fetch invocation?

@divijvaidya To pre-create the valid index file in remote storage cache dir , I initially fetched it from the remote storage and then run the scenario. The 1 count is for pre-processing.

Thanks for the explanation. So that will confuse readers. Please do either:

reset the mock rsm
or

run verifyFetchIndexInvocation(count = 1) again before L672, so that we can make sure the count is not increasing even we call cache.getIndexEntry again.

@iit2009060 , have you addressed this comment?

Hey @showuon Thanks for identifying it , I missed it.
Changes Done.

divijvaidya · 2023-10-04T08:08:15Z

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

+        // The size of the string written in the file is 12 bytes,
+        // but it should be multiple of Offset Index EntrySIZE which is equal to 8.
+        pw.close()
+        val offsetIdx = createOffsetIndexForSegmentMetadata(metadata)


there is a bug in existing createOffsetIndexForSegmentMetadata and similar function. They create cache at the wrong place in the line:
new OffsetIndex(remoteOffsetIndexFile(tpDir, metadata)
Instead of tpDir, it should point to remote cache director which is a folder inside tpDir.

Please fix that and see if it impacts your test in any way.

@divijvaidya But when we call get index entry , it will create indexes file under remote cache directory only. It will read file content from tpDir and create indexes in "remoteIndexCache" dir.
IMO "createOffsetIndexForSegmentMetadata" will create indexes file and tpDir act as a placeholder for files fetched from remoteStorage. If we change it to remote cache dir , all test cases will always have the file exist behaviour.

@divijvaidya We can keep this function i.e createOffsetIndexForSegmentMetadata to be used for indexes received from remote storage rather than remoteindexcache.

The existing test case are working because they directly put values into internalCache(Caeffine) rather than the getIndexEntry route.

In the above test case I want to corrupt the Index files received from remote storage rather than from remote index cache.
getIndexEntry does following steps

Fetch indexes from remote Storage( i.e indexes stored in tpDir directory as per the mock)

Manually Corrupt the indexes file stored in tpDir Directory after fetch from remote storage.

It tries to copy indexes files in tpDir directory to remote index cache directory.

Run sanity check on indexes stored in tpDir/remoteindexcache directory.

It throws CorruptedIndexException.

divijvaidya · 2023-10-04T08:09:32Z

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

+        // The size of the string written in the file is 12 bytes,
+        // but it should be multiple of Offset Index EntrySIZE which is equal to 8.
+        pw.close()
+        val offsetIdx = createOffsetIndexForSegmentMetadata(metadata)


could we have 3 variants of this test where we are corruption all 3 indexes one by one. You can use @Parameterized to run those scenarios in same test.

@divijvaidya Is it possible to parametrize based on function as a parameter ? Otherwise I need to write if else condition in the same test ?
Corrupting Time and Offset Index is easy based on entry_size , any suggestion how to do it for TransactionIndex ?

@iit2009060 , if / else is fine with me, just make the test clear.

For txnIndex corruption, I usually check other tests if they did something similar as we want. Here's a good reference:

kafka/core/src/test/scala/unit/kafka/log/TransactionIndexTest.scala

Lines 62 to 73 in fbc39bc

def testSanityCheck(): Unit = {

val abortedTxns = List(

new AbortedTxn(0L, 0, 10, 11),

new AbortedTxn(1L, 5, 15, 13),

new AbortedTxn(2L, 18, 35, 25),

new AbortedTxn(3L, 32, 50, 40))

abortedTxns.foreach(index.append)

index.close()

// open the index with a different starting offset to fake invalid data

val reopenedIndex = new TransactionIndex(100L, file)

assertThrows(classOf[CorruptIndexException], () => reopenedIndex.sanityCheck())

@divijvaidya @showuon done

divijvaidya · 2023-10-04T08:11:59Z

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

+    // wait until entry is marked for deletion
+    TestUtils.waitUntilTrue(() => entry.isMarkedForCleanup,
+      "Failed to mark cache entry for cleanup after invalidation")
+    TestUtils.waitUntilTrue(() => entry.isCleanStarted,


Clean started doesn't mean that index has been deleted. There could be a case where next line executes even though file has not been deleted yet. I would suggest to wrap the next assertions in waitUntilTrue(). It would make the test non-flaky.

showuon

@iit2009060 , thanks for the PR. Left some comments.

showuon · 2023-10-05T03:33:42Z

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

+
+    // restore index files
+    renameRemoteCacheIndexFileFromDisk(tempSuffix)
+    // validate cache entry for the above key should be  null


nit: additional space before null.

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

showuon · 2023-10-05T05:07:52Z

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

+        // The size of the string written in the file is 12 bytes,
+        // but it should be multiple of Offset Index EntrySIZE which is equal to 8.
+        pw.close()
+        val offsetIdx = createOffsetIndexForSegmentMetadata(metadata)


@iit2009060 , if / else is fine with me, just make the test clear.

For txnIndex corruption, I usually check other tests if they did something similar as we want. Here's a good reference:

kafka/core/src/test/scala/unit/kafka/log/TransactionIndexTest.scala

Lines 62 to 73 in fbc39bc

def testSanityCheck(): Unit = {

val abortedTxns = List(

new AbortedTxn(0L, 0, 10, 11),

new AbortedTxn(1L, 5, 15, 13),

new AbortedTxn(2L, 18, 35, 25),

new AbortedTxn(3L, 32, 50, 40))

abortedTxns.foreach(index.append)

index.close()

// open the index with a different starting offset to fake invalid data

val reopenedIndex = new TransactionIndex(100L, file)

assertThrows(classOf[CorruptIndexException], () => reopenedIndex.sanityCheck())

showuon · 2023-10-05T12:06:32Z

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

+    // Index  Files already exist
+    // rsm should not be called again
+    // instead files exist on disk
+    // should be used


Please make these 4 lines into 2.

showuon · 2023-10-05T12:09:20Z

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

+      })
+    cache.getIndexEntry(rlsMetadata)
+    // No exception should occur
+    // Offset rsm 0,TimeIndex 1 , TxnIndex 1


What is offset rsm? OffsetIndex?

Updated the description. Yes it was OffsetIndex

showuon · 2023-10-05T12:10:17Z

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

+    // However cache fetch failed
+    // but it has  already created offset and time index file in remote cache dir.
+    // Current status
+    // ( cache  is null)
+    // RemoteCacheDir contain
+    // 1. Offset Index File which is fine and not corrupted
+    // 2. Time Index File which is corrupted
+    // What should be the code flow in next execution
+    // 1. No rsm call for fetching OffSet Index File.
+    // 2. Time index file should be fetched from remote storage again as it is corrupted in the first execution.


Please remove addition space between words.

showuon · 2023-10-05T12:11:02Z

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

+    // 1. Offset Index File which is fine and not corrupted
+    // 2. Time Index File which is corrupted


The word which can be removed:
// 1. Offset Index File is fine and not corrupted
// 2. Time Index File is corrupted

divijvaidya · 2023-10-05T13:50:12Z

@showuon @iit2009060 please don't wait for my review on this one. I might not get to it until next week.

iit2009060 · 2023-10-06T04:48:59Z

@showuon @iit2009060 please don't wait for my review on this one. I might not get to it until next week.

@showuon Can you review @divijvaidya comments. I have taken care of all the reviews.

showuon

@iit2009060 , overall LGTM, left some refactor comments.

showuon · 2023-10-06T07:11:24Z

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

+  @ParameterizedTest
+  @EnumSource(value = classOf[IndexType], names = Array("OFFSET", "TIMESTAMP", "TRANSACTION"))
+  def testCorruptCacheIndexFileExistsButNotInCache(indexType: IndexType): Unit = {
+    // create Corrupt  Index File in remote index cache


// create Corrupted Index File in remote index cache

format is wrong:

if (indexType == IndexType.OFFSET) { createCorruptOffsetIndexFile(cache.cacheDir()) } else if (indexType == IndexType.TIMESTAMP) { createCorruptTimeIndexOffsetFile(cache.cacheDir()) } else if (indexType == IndexType.TRANSACTION) { createCorruptTxnIndexForSegmentMetadata(cache.cacheDir(), rlsMetadata) }

showuon · 2023-10-06T07:14:44Z

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

    // assert that parent directory for the index files is correct
    assertEquals(RemoteIndexCache.DIR_NAME, offsetIndexFile.getParent.getFileName.toString,
-      s"offsetIndex=$offsetIndexFile is not overwrite under incorrect parent")
+      s"offsetIndex=entry.offsetIndex().file().toPath is created under incorrect parent")


Should be s"offsetIndex=$offsetIndexFile is created under incorrect parent") ?

showuon · 2023-10-06T07:15:36Z

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

+        val timeIdx = createTimeIndexForSegmentMetadata(metadata)
+        val txnIdx = createTxIndexForSegmentMetadata(metadata)
+        maybeAppendIndexEntries(offsetIdx, timeIdx)
+        // Create corrupt index file which would be returned by rsm


// Create corrupt index file which would be returned by rsm
->
// Create corrupted index file

showuon · 2023-10-06T07:20:21Z

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

+    // However cache fetch failed
+    // but it has already created offset and time index file in remote cache dir.


These 2 lines can be removed.

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

showuon · 2023-10-06T07:24:02Z

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

+    // No exception should occur
+    // rsm should not be called for offset Index


// No exception should occur <- should be removed
// rsm should not be called for offset Index -> // rsm should not be called to fetch offset Index

showuon · 2023-10-06T07:43:25Z

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

+    // Index  Files already exist ,rsm should not be called again
+    // instead files exist on disk should be used


// Index Files already exist, rsm should not fetch them again.

and remove L654

showuon · 2023-10-06T07:44:59Z

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

+    // rsm should not be called again
+    // instead files exist on disk
+    // should be used
+    verifyFetchIndexInvocation(count = 1)


Thanks for the explanation. So that will confuse readers. Please do either:

reset the mock rsm
or

run verifyFetchIndexInvocation(count = 1) again before L672, so that we can make sure the count is not increasing even we call cache.getIndexEntry again.

showuon · 2023-10-06T07:46:04Z

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

+        if (testIndexType == IndexType.OFFSET) {
+          createCorruptOffsetIndexFile(tpDir)
+        }
+        else if (testIndexType == IndexType.TIMESTAMP) {
+          createCorruptTimeIndexOffsetFile(tpDir)
+        }
+        else if (testIndexType == IndexType.TRANSACTION) {
+          txnIdx = createCorruptTxnIndexForSegmentMetadata(tpDir, rlsMetadata)
+        }


format. Thanks.

And since these logic repeat again, we can extract them into another helper method. Ex:

private void createCorruptedIndexFile(IndexType type) { if (testIndexType == IndexType.OFFSET) { createCorruptOffsetIndexFile(tpDir) ..... }

Then, in the test, we can just call createCorruptedIndexFile(testIndexType) to do it.

iit2009060 · 2023-10-06T08:21:36Z

@iit2009060 , overall LGTM, left some refactor comments.

@showuon done as mentioned.

showuon

LGTM! Just a minor formatting comment.

showuon · 2023-10-06T09:49:22Z

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

+  @EnumSource(value = classOf[IndexType], names = Array("OFFSET", "TIMESTAMP", "TRANSACTION"))
+  def testCorruptCacheIndexFileExistsButNotInCache(indexType: IndexType): Unit = {
+    // create Corrupted Index File in remote index cache
+    createCorruptedIndexFile(indexType,cache.cacheDir())


nit: space after comma (indexType, cache.cacheDir()

Thanks for the quick fix. Let's wait for the CI build completes.

@showuon Unrelated test failures.
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/9/tests

kamalcph

Since we are adding new tests, can we create a new RemoteIndexCacheTest under storage module and start writing the tests in java?

The source code is in Java but the test is in scala. We can move the existing tests in a separate PR.

kamalcph · 2023-10-09T04:33:28Z

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

@@ -166,6 +166,11 @@ public Cache<Uuid, Entry> internalCache() {
        return internalCache;
    }

+    // Visible for testing
+    public File cacheDir() {


Can we file a ticket to move the RemoteIndexCacheTest to storage module under the same package?

@kamalcph I have just started understanding of the kafka ecosystem , Can you help me understand the rational behind it , Then i can create a ticket with details.

You can find more details about the intention to move the source code to Java in KAFKA-14524

iit2009060 · 2023-10-09T04:59:22Z

Since we are adding new tests, can we create a new RemoteIndexCacheTest under storage module and start writing the tests in java?

The source code is in Java but the test is in scala. We can move the existing tests in a separate PR.

@kamalcph Can i create a ticket for same and work on it after PR merge. This has been pending from last week and already approved.

Since we are adding new tests, can we create a new RemoteIndexCacheTest under storage module and start writing the tests in java?

The source code is in Java but the test is in scala. We can move the existing tests in a separate PR.

iit2009060 · 2023-10-09T05:02:22Z

@showuon @kamalcph @divijvaidya I have rebased changes and resolved conflict on the changes merged by
the PR #14381.
Can you please review and merge it.

kamalcph

Changes in the patch, LGTM. We can rewrite it to Java in a separate patch.

kamalcph · 2023-10-09T04:44:35Z

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

+  @EnumSource(value = classOf[IndexType], names = Array("OFFSET", "TIMESTAMP", "TRANSACTION"))
+  def testCorruptCacheIndexFileExistsButNotInCache(indexType: IndexType): Unit = {
+    // create Corrupted Index File in remote index cache
+    createCorruptedIndexFile(indexType, cache.cacheDir())
    val entry = cache.getIndexEntry(rlsMetadata)
    // Test would fail if it throws corrupt Exception


Can we update this comment to?

Test would fail it it throws exception other than CorruptIndexException

kamalcph · 2023-10-09T05:01:24Z

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

+    // rsm should not be called to fetch offset Index
+    verifyFetchIndexInvocation(0, Seq(IndexType.OFFSET))
+    verifyFetchIndexInvocation(1, Seq(IndexType.TIMESTAMP))
+    verifyFetchIndexInvocation(1, Seq(IndexType.TRANSACTION))


Txn index file was not corrupted. Can we add a comment to explain why are we fetching it again?

showuon · 2023-10-09T05:25:15Z

@iit2009060 , there are compilation failure, could you fix it?
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/10/pipeline/

iit2009060 · 2023-10-09T05:31:09Z

@iit2009060 , there are compilation failure, could you fix it? https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/10/pipeline/

on it.

iit2009060 · 2023-10-09T08:00:42Z

@hudeqi @showuon There is one test case failing related to the above change #14381, can you help me resolving it , in the local it is not failing ?
Build / JDK 8 and Scala 2.12 / testClearCacheAndIndexFilesWhenResizeCache() – kafka.log.remote.RemoteIndexCacheTest
java.io.UncheckedIOException: java.nio.file.NoSuchFileException: /tmp/kafka-RemoteIndexCacheTest5502978390891743756/mwbAzcjOTtKRIDxlQrgTTw:foo-0/remote-log-index-cache/0_xg9JTaPnS4C47HfCgv2_gA.txnindex.deleted
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/12/tests/

hudeqi · 2023-10-09T08:29:49Z

@hudeqi @showuon There is one test case failing related to the above change #14381, can you help me resolving it , in the local it is not failing ? Build / JDK 8 and Scala 2.12 / testClearCacheAndIndexFilesWhenResizeCache() – kafka.log.remote.RemoteIndexCacheTest java.io.UncheckedIOException: java.nio.file.NoSuchFileException: /tmp/kafka-RemoteIndexCacheTest5502978390891743756/mwbAzcjOTtKRIDxlQrgTTw:foo-0/remote-log-index-cache/0_xg9JTaPnS4C47HfCgv2_gA.txnindex.deleted https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/12/tests/

hi, I think this may be the reason: when executing getIndexFileFromRemoteCacheDir, the .deleted file will be listed in the filter(Files.isRegularFile(_)) stage, but when executing path.getFileName.toString, the .deleted file has already been deleted because cleanStarted is already true to start cleaning.
If possible, can I fix this issue in your branch? @iit2009060

iit2009060 · 2023-10-09T08:32:00Z

@hudeqi @showuon There is one test case failing related to the above change #14381, can you help me resolving it , in the local it is not failing ? Build / JDK 8 and Scala 2.12 / testClearCacheAndIndexFilesWhenResizeCache() – kafka.log.remote.RemoteIndexCacheTest java.io.UncheckedIOException: java.nio.file.NoSuchFileException: /tmp/kafka-RemoteIndexCacheTest5502978390891743756/mwbAzcjOTtKRIDxlQrgTTw:foo-0/remote-log-index-cache/0_xg9JTaPnS4C47HfCgv2_gA.txnindex.deleted https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/12/tests/

hi, I think this may be the reason: when executing getIndexFileFromRemoteCacheDir, the .deleted file will be listed in the filter(Files.isRegularFile(_)) stage, but when executing path.getFileName.toString, the .deleted file has already been deleted because cleanStarted is already true to start cleaning. If possible, can I fix this issue in your branch? @iit2009060

Yes sure @hudeqi . This seems to be a case of flaky behaviour i.e. race condition ?

hudeqi · 2023-10-09T09:08:22Z

@hudeqi @showuon There is one test case failing related to the above change #14381, can you help me resolving it , in the local it is not failing ? Build / JDK 8 and Scala 2.12 / testClearCacheAndIndexFilesWhenResizeCache() – kafka.log.remote.RemoteIndexCacheTest java.io.UncheckedIOException: java.nio.file.NoSuchFileException: /tmp/kafka-RemoteIndexCacheTest5502978390891743756/mwbAzcjOTtKRIDxlQrgTTw:foo-0/remote-log-index-cache/0_xg9JTaPnS4C47HfCgv2_gA.txnindex.deleted https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/12/tests/

hi, I think this may be the reason: when executing getIndexFileFromRemoteCacheDir, the .deleted file will be listed in the filter(Files.isRegularFile(_)) stage, but when executing path.getFileName.toString, the .deleted file has already been deleted because cleanStarted is already true to start cleaning. If possible, can I fix this issue in your branch? @iit2009060

Yes sure @hudeqi . This seems to be a case of flaky behaviour i.e. race condition ?

I think so, please cherry-pick this commit: hudeqi@854c773, thanks. @iit2009060

iit2009060 · 2023-10-09T09:28:12Z

@hudeqi @showuon There is one test case failing related to the above change #14381, can you help me resolving it , in the local it is not failing ? Build / JDK 8 and Scala 2.12 / testClearCacheAndIndexFilesWhenResizeCache() – kafka.log.remote.RemoteIndexCacheTest java.io.UncheckedIOException: java.nio.file.NoSuchFileException: /tmp/kafka-RemoteIndexCacheTest5502978390891743756/mwbAzcjOTtKRIDxlQrgTTw:foo-0/remote-log-index-cache/0_xg9JTaPnS4C47HfCgv2_gA.txnindex.deleted https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/12/tests/

hi, I think this may be the reason: when executing getIndexFileFromRemoteCacheDir, the .deleted file will be listed in the filter(Files.isRegularFile(_)) stage, but when executing path.getFileName.toString, the .deleted file has already been deleted because cleanStarted is already true to start cleaning. If possible, can I fix this issue in your branch? @iit2009060

Yes sure @hudeqi . This seems to be a case of flaky behaviour i.e. race condition ?

I think so, please cherry-pick this commit: hudeqi@854c773, thanks. @iit2009060

@hudeqi This is given me bad object/revision . Can you have the command handy to do cherry pick from another fork ?
git cherry-pick 854c773
fatal: bad object 854c773
This seems to be existing in your fork repository
I figured out through this link. Will update soon the PR with changes.

iit2009060 · 2023-10-09T09:34:50Z

@hudeqi @showuon There is one test case failing related to the above change #14381, can you help me resolving it , in the local it is not failing ? Build / JDK 8 and Scala 2.12 / testClearCacheAndIndexFilesWhenResizeCache() – kafka.log.remote.RemoteIndexCacheTest java.io.UncheckedIOException: java.nio.file.NoSuchFileException: /tmp/kafka-RemoteIndexCacheTest5502978390891743756/mwbAzcjOTtKRIDxlQrgTTw:foo-0/remote-log-index-cache/0_xg9JTaPnS4C47HfCgv2_gA.txnindex.deleted https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/12/tests/

hi, I think this may be the reason: when executing getIndexFileFromRemoteCacheDir, the .deleted file will be listed in the filter(Files.isRegularFile(_)) stage, but when executing path.getFileName.toString, the .deleted file has already been deleted because cleanStarted is already true to start cleaning. If possible, can I fix this issue in your branch? @iit2009060

Yes sure @hudeqi . This seems to be a case of flaky behaviour i.e. race condition ?

I think so, please cherry-pick this commit: hudeqi@854c773, thanks. @iit2009060

@hudeqi This is given me bad object/revision . Can you have the command handy to do cherry pick from another fork ? git cherry-pick 854c773 fatal: bad object 854c773 This seems to be existing in your fork repository I figured out through this link. Will update soon the PR with changes.

@hudeqi @showuon Please review it

iit2009060 · 2023-10-09T12:23:12Z

Unrelated test failures
https://ci-builds.apache.org/blue/organizations/jenkins/Kafka%2Fkafka-pr/detail/PR-14482/14/tests
@showuon @divijvaidya can we merge this changes ?

iit2009060 · 2023-10-10T03:56:34Z

@showuon Can we merge this changes ?

showuon · 2023-10-10T05:17:37Z

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

+    TestUtils.waitUntilTrue(() => cacheEntry.isCleanFinished,
+      "Failed to finish cleanup cache entry after resizing cache.")

    // verify no index files on remote cache dir
    TestUtils.waitUntilTrue(() => !getIndexFileFromRemoteCacheDir(LogFileUtils.INDEX_FILE_SUFFIX).isPresent,


@iit2009060 @hudeqi , adding an isCleanFinished flag just for test is not a good solution. In this case, could we catch the exception in the getIndexFileFromRemoteCacheDir?

@hudeqi Can you resolve the above comment ?

@showuon I have removed the isCleanFinished flag changes and catch the Exception in the method.
cc @hudeqi
Please review it .

showuon

LGTM! Thanks for the update.

showuon · 2023-10-11T02:56:52Z

Failed tests are unrelated and also failed in trunk build.

est Cases Covered 1. Index Files already exist on disk but not in Cache i.e. RemoteIndexCache should not call remoteStorageManager to fetch it instead cache it from the local index file present. 2. RSM returns CorruptedIndex File i.e. RemoteIndexCache should throw CorruptedIndexException instead of successfull execution. 3. Deleted Suffix Indexes file already present on disk i.e. If cleaner thread is slow , then there is a chance of deleted index files present on the disk while in parallel same index Entry is invalidated. To understand more refer https://issues.apache.org/jira/browse/KAFKA-15169 Reviewers: Divij Vaidya <diviv@amazon.com>, Luke Chen <showuon@gmail.com>, Kamal Chandraprakash<kamal.chandraprakash@gmail.com>

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

divijvaidya reviewed Oct 4, 2023

View reviewed changes

showuon reviewed Oct 5, 2023

View reviewed changes

iit2009060 force-pushed the KAFKA-15169 branch from 99bf0ad to 220dfb6 Compare October 5, 2023 11:54

showuon reviewed Oct 5, 2023

View reviewed changes

showuon reviewed Oct 6, 2023

View reviewed changes

iit2009060 force-pushed the KAFKA-15169 branch from 4edf149 to db648c1 Compare October 6, 2023 08:54

showuon approved these changes Oct 6, 2023

View reviewed changes

kamalcph reviewed Oct 9, 2023

View reviewed changes

iit2009060 force-pushed the KAFKA-15169 branch from 683ab3a to f953439 Compare October 9, 2023 04:55

kamalcph approved these changes Oct 9, 2023

View reviewed changes

iit2009060 force-pushed the KAFKA-15169 branch from f953439 to 951decd Compare October 9, 2023 05:24

iit2009060 added 7 commits October 9, 2023 14:53

Added TestCase in RemoteIndexCache

c8bac68

Use waitUntilTrue,correct Indentation

19c1134

Parametrized all corrupt index file cases

f43bd00

Adding test case of reproducing corrupt exception using code flow

980a233

correcting spaces and wording

732af54

Moving corrupt index file creation into a generic function

3b4569c

correcting spaces and wording

074405b

Added comments and description

414bcbb

iit2009060 force-pushed the KAFKA-15169 branch from cfe4ee5 to 414bcbb Compare October 9, 2023 09:24

fix flaky test in RemoteIndexCacheTest

b1f1621

showuon reviewed Oct 10, 2023

View reviewed changes

Catch FileNotFoundException in resize cache test

cdb2391

showuon approved these changes Oct 11, 2023

View reviewed changes

showuon merged commit 99ce2e0 into apache:trunk Oct 11, 2023
1 check failed

		def testOffsetIndexFileAlreadyExistOnDiskButNotInCache(): Unit ={
		val remoteIndexCacheDir = new File(tpDir,RemoteIndexCache.DIR_NAME)

	def testSanityCheck(): Unit = {
	val abortedTxns = List(
	new AbortedTxn(0L, 0, 10, 11),
	new AbortedTxn(1L, 5, 15, 13),
	new AbortedTxn(2L, 18, 35, 25),
	new AbortedTxn(3L, 32, 50, 40))
	abortedTxns.foreach(index.append)
	index.close()

	// open the index with a different starting offset to fake invalid data
	val reopenedIndex = new TransactionIndex(100L, file)
	assertThrows(classOf[CorruptIndexException], () => reopenedIndex.sanityCheck())

		// 1. Offset Index File which is fine and not corrupted
		// 2. Time Index File which is corrupted

		// However cache fetch failed
		// but it has already created offset and time index file in remote cache dir.

		// No exception should occur
		// rsm should not be called for offset Index

		// Index Files already exist ,rsm should not be called again
		// instead files exist on disk should be used

KAFKA-15169: Added TestCase in RemoteIndexCache #14482

KAFKA-15169: Added TestCase in RemoteIndexCache #14482

Conversation

iit2009060 commented Oct 4, 2023 • edited

Test Cases Covered

Committer Checklist (excluded from commit message)

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

iit2009060 Oct 4, 2023 • edited

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

iit2009060 Oct 4, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iit2009060 Oct 4, 2023 • edited

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

showuon 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

Choose a reason for hiding this comment

iit2009060 Oct 5, 2023 • edited

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

divijvaidya commented Oct 5, 2023

iit2009060 commented Oct 6, 2023 • edited

showuon 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iit2009060 commented Oct 6, 2023

showuon 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

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

iit2009060 commented Oct 9, 2023

iit2009060 commented Oct 9, 2023

kamalcph left a comment

iit2009060 commented Oct 4, 2023 •

edited

iit2009060 Oct 4, 2023 •

edited

iit2009060 Oct 4, 2023 •

edited

iit2009060 Oct 4, 2023 •

edited

iit2009060 Oct 5, 2023 •

edited

iit2009060 commented Oct 6, 2023 •

edited

iit2009060 commented Oct 9, 2023 •

edited