You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jeel2420 (via GitHub)" <gi...@apache.org> on 2023/10/04 10:01:03 UTC
[PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
jeel2420 opened a new pull request, #14483:
URL: https://github.com/apache/kafka/pull/14483
This PR fixes the concurrency bug in RemoteIndexCache.
(From Jira description)
RemoteIndexCache has a concurrency bug which leads to IOException while fetching data from remote tier.
Below events in order of timeline -
- Thread 1 (cache thread): invalidates the entry, removalListener is invoked async, so the files have not been renamed to "deleted" suffix yet.
- Thread 2: (fetch thread): tries to find entry in cache, doesn't find it because it has been removed by 1, fetches the entry from S3, writes it to existing file (using replace existing)
- Thread 1: async removalListener is invoked, acquires a lock on old entry (which has been removed from cache), it renames the file to "deleted" and starts deleting it
- Thread 2: Tries to create in-memory/mmapped index, but doesn't find the file and hence, creates a new file of size 2GB in AbstractIndex constructor. JVM returns an error as it won't allow creation of 2GB random access file.
**Fix**: Used `EvictionListener` instead of `RemovalListener` to perform the eviction synchronously in Caffeine cache and for the manual removal used computeIfAbsent to rename and delete the key from the cache synchronously by returning null.
Added `testConcurrentRemoveReadForCache` to reproduce the bug by following the above timeline of events. Unit test case is passing now.
Jira
https://issues.apache.org/jira/browse/KAFKA-15481
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1347078859
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -538,7 +533,75 @@ class RemoteIndexCacheTest {
assertEquals(RemoteIndexCache.DIR_NAME, offsetIndexFile.getParent.getFileName.toString,
s"offsetIndex=$offsetIndexFile is not overwrite under incorrect parent")
// file is corrupted it should fetch from remote storage again
- verifyFetchIndexInvocation(count = 1)
+ verifyFetchIndexInvocation(count = 1, indexTypes = Seq(IndexType.OFFSET))
+ }
+
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ } else {
+ // Subsequent call for markForCleanup method
+ latchForCacheRead.countDown()
+ latchForCacheRemove.countDown()
+ }
Review Comment:
I found markForCleanup getting called at the end of the test maybe by cleanup() (not sure). But after that I added latchForTestWait so this might not be required anymore. I will check and remove if it is not required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1382994668
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
+ }
+ } catch (IOException e) {
Review Comment:
Fair point. Please create a JIRA for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364998096
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@showuon
`For (1), the markForCleanUp is an injected method for controlling the invoking order. So there are latches wait/countdown.`
Do you mean this is a mock method and no rename would happen in this case ?
Effectively the functionality/logic of `markCleanUp` is called one time only ?
I was thinking something like this
```
`val latchForTestWait = new CountDownLatch(2)
val removeCache = (() => {
cache.remove(rlsMetadata.remoteLogSegmentId().id())
latchForTestWait.countdown()
}): Runnable
val readCache = (() => {
entry = cache.getIndexEntry(rlsMetadata)
// Signal the CacheRemove to start renaming the files
latchForTestWait.countDown()
}): Runnable
val executor = Executors.newFixedThreadPool(2)
try {
executor.submit(removeCache: Runnable)
executor.submit(readCache: Runnable)
// Wait for signal to complete the test
latchForTestWait.await()
// validate cache size `
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363504542
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 `markForCleanUp` should be called only one time. We need to test the behaviour when there are concurrent read/remove happens on the cache for the same entry.
In the test we just need to assert the way @showuon suggested
// So, maybe we verify with this:
if (Files.exists(entry.offsetIndex().file().toPath)) {
assertCacheSize(1)
} else {
assertCacheSize(0)
}
Calling '`markForCleanUp`' twice will always result in cacheSize 0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1360317032
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
> 3. readCache executed and finished because no lock is pending on the remove operation
No, it's not right. The lock is in the concurrentMap inside cache.
But checking it again, I found the lock is when "updating the entry". That is, we have 2 threads trying to do:
1. create a new entry for the key
2. remove the entry for the key
We can't make sure which one will complete first. All we can make sure is the 2 operations are atomic. So, after L646:
```
// Wait for signal to complete the test
latchForTestWait.await()
// Here, we can't make sure if the cache size is 0 or 1 at this point. I think we can only do is to make sure either it's cache size 1 and file existed (create goes later), or cache size is 0 and file inexisted (remove goes later)
// So, maybe we verify with this:
if (Files.exists(entry.offsetIndex().file().toPath)) {
assertCacheSize(1)
} else {
assertCacheSize(0)
}
```
WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1360248436
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
> I was just thinking around it , if this fails , it will create a dump of files with delete suffix entry which never gets deleted from the disk ? Is the behaviour ok ?
It's OK, it will be deleted in next startup. Of course we can have a way to retry enqueue them, but I think it's a pretty rare case, we can keep it as is.
> During remove we try to take a readLock but in markForCleanUp we try to take a write lock , Will it not result a deadlock state ?
The readlock in `remove` method is `RemoteIndexCache#lock`. The writeLock in `markForCleanUp` method is `Entry#lock`. I agree it's confusing to have the same name within the same parent class. Could you file another PR to update the lock variable name?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363504542
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 `markForCleanUp` should be called only one time. We need to test the behaviour when there are concurrent read/remove happens on the cache for the same entry.
In the test we just need to assert the way @showuon suggested
// So, maybe we verify with this:
if (Files.exists(entry.offsetIndex().file().toPath)) {
assertCacheSize(1)
} else {
assertCacheSize(0)
}
Calling 'markForCleanUp' twice asserting CacheSize to be zero which is wrong.
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 `markForCleanUp` should be called only one time. We need to test the behaviour when there are concurrent read/remove happens on the cache for the same entry.
In the test we just need to assert the way @showuon suggested
// So, maybe we verify with this:
if (Files.exists(entry.offsetIndex().file().toPath)) {
assertCacheSize(1)
} else {
assertCacheSize(0)
}
Calling '`markForCleanUp`' twice asserting CacheSize to be zero which is wrong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1365036479
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
Yes correct , it is testing the same thing. I am also fine. But from readability perspective the one I propose is simpler to understand. I left @jeel2420 to make a decision here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1772367941
Thank you @showuon @iit2009060 for your review. I will go ahead and merge this PR now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1366804550
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -151,9 +151,8 @@ private Cache<Uuid, Entry> initEmptyCache(long maxSize) {
.weigher((Uuid key, Entry entry) -> {
return (int) entry.entrySizeBytes;
})
- // removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
- // evicted (means removal due to the policy)
- .removalListener((Uuid key, Entry entry, RemovalCause cause) -> {
+ // evictionListener is invoked when RemovalCause.wasEvicted() is true
Review Comment:
Please update the comment here to explicitly specify whether it would be executed sync or async with cache eviction. You can say:
This listener is invoked each time an entry is being automatically removed due to eviction. The cache with invoke this listener during the atomic operation to remove the entry (refer: https://github.com/ben-manes/caffeine/wiki/Removal), hence, care must be taken to ensure that this operation is not expensive.
Note that this listener is not invoked when RemovalCause from cache is EXPLICIT or REPLACED (e.g. on Cache.invalidate(), Cache.put() etc.) For a complete list see: https://www.javadoc.io/static/com.github.ben-manes.caffeine/caffeine/2.9.2/index.html?com/github/benmanes/caffeine/cache/Caffeine.html. Hence, any operation required after removal from cache must be performed manually for these scenarios.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -187,7 +186,18 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ if (!expiredIndexes.offer(v)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", v, k);
+ }
+ } catch (IOException e) {
+ throw new KafkaException(e);
+ }
+ // Returning null to remove the key from the cache
+ return null;
+ });
Review Comment:
could be moved into a function since this is used 3 times in the same class in remove(), in removeAll() and in evictionListener.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1378860641
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
+ }
+ } catch (IOException e) {
Review Comment:
I am not sure the I/O exception is handled correctly here i.e. follows the convention for I/O error handling in Kafka. Maybe the log dir failure channel should be triggered?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
Review Comment:
What is the max capacity set for the queue? Also isn't there a problem is an entry is removed from the cache, marked for clean-up but never added to the expiration queue as the corresponding index files would be leaked on the file system?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1346940422
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -169,7 +169,16 @@ public Cache<Uuid, Entry> internalCache() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ v.cleanup();
Review Comment:
`markForCleanup` is enough, it'll set the cleanup flag as true, and rename files. `cleanup` can be done in a cleanup thread.
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -178,7 +187,16 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363730045
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ entry = cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ if (getIndexFileFromRemoteCacheDir(cache, LogFileUtils.INDEX_FILE_SUFFIX).isPresent) {
+ assertCacheSize(1)
+ } else {
+ assertCacheSize(0)
+ }
Review Comment:
Added a comment before assertion to explain both cases (Read thread go first and Remove thread go first).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1362060707
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
Thank you @showuon @iit2009060 for your valuable comments. I will update the PR with the changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1362343057
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
Using `expiredIndexes.offer` now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1348624218
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -568,27 +621,26 @@ class RemoteIndexCacheTest {
}
private def verifyFetchIndexInvocation(count: Int,
- indexTypes: Seq[IndexType] =
- Seq(IndexType.OFFSET, IndexType.TIMESTAMP, IndexType.TRANSACTION)): Unit = {
+ indexTypes: Seq[IndexType]): Unit = {
for (indexType <- indexTypes) {
verify(rsm, times(count)).fetchIndex(any(classOf[RemoteLogSegmentMetadata]), ArgumentMatchers.eq(indexType))
}
}
private def createTxIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TransactionIndex = {
- val txnIdxFile = remoteTransactionIndexFile(tpDir, metadata)
+ val txnIdxFile = remoteTransactionIndexFile(new File(tpDir, DIR_NAME), metadata)
txnIdxFile.createNewFile()
new TransactionIndex(metadata.startOffset(), txnIdxFile)
}
private def createTimeIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TimeIndex = {
val maxEntries = (metadata.endOffset() - metadata.startOffset()).asInstanceOf[Int]
- new TimeIndex(remoteTimeIndexFile(tpDir, metadata), metadata.startOffset(), maxEntries * 12)
+ new TimeIndex(remoteTimeIndexFile(new File(tpDir, DIR_NAME), metadata), metadata.startOffset(), maxEntries * 12)
Review Comment:
Make sense. Reverted the change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1359202415
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
Do something like we did in the other place:
```
if (!expiredIndexes.offer(entry)) {
log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1776771107
@hudeqi
> > > @jeel2420 https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14483/16/testReport/junit/kafka.log.remote/RemoteIndexCacheTest/Build___JDK_8_and_Scala_2_12___testClearCacheAndIndexFilesWhenResizeCache__/ test is failing in CI. Please check the reason.
> >
> >
> > I found that it’s a flaky test case. It fails sometimes when I tested locally as well. I will take look into it.
>
> I have fixed it in trunk before, maybe this pr need to merge latest trunk and it will pass. @jeel2420
Hey @hudeqi
This is failing in trunk as well and not just this PR. I have added details in https://issues.apache.org/jira/browse/KAFKA-15671 From a quick look seems like we are catching NoSuchFileFound whereas it's being thrown as a root cause and not a top level exception. Since you are familiar, please help guide Brian (who has volunteered to fix this ticket) towards a fix. It would be great if you would review their code once they have a PR open.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1345548349
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -132,7 +133,7 @@ class RemoteIndexCacheTest {
// this call should have invoked fetchOffsetIndex, fetchTimestampIndex once
val resultPosition = cache.lookupOffset(rlsMetadata, offsetPosition1.offset)
assertEquals(offsetPosition1.position, resultPosition)
- verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET, IndexType.TIMESTAMP))
+ verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET))
Review Comment:
Checking fetchIndex invocation for Offset index type only as fetchIndex mock is creating all index files during the first invocation so for remaining index types there will be no fetchIndex Invocation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1775107621
Adding here the reason @jeel2420 mentioned the cause of flaky test:
```
In getIndexFileFromRemoteCacheDir function we basically run below code to get the files from the disk and checking if file with specific suffix is present or not.
Files.walk(cache.cacheDir().toPath())
.filter(Files.isRegularFile(_))
.filter(path => path.getFileName.toString.endsWith(suffix))
.findAny()
Here in cacheDir after the remove call we can have files with .deleted suffix and those files can be removed from cacheDir at any time. So if .deleted suffix file got deleted in between when the above code is running which basically traverse the directory structure we can get UncheckedIOException which is not handled in the function.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "hudeqi (via GitHub)" <gi...@apache.org>.
hudeqi commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1776392320
> > @jeel2420 https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14483/16/testReport/junit/kafka.log.remote/RemoteIndexCacheTest/Build___JDK_8_and_Scala_2_12___testClearCacheAndIndexFilesWhenResizeCache__/ test is failing in CI. Please check the reason.
>
> I found that it’s a flaky test case. It fails sometimes when I tested locally as well. I will take look into it.
I have fixed it in trunk before, maybe this pr need to merge latest trunk and it will pass. @jeel2420
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1358347561
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -137,13 +137,10 @@ public RemoteIndexCache(long maxSize, RemoteStorageManager remoteStorageManager,
public void resizeCacheSize(long remoteLogIndexFileCacheSize) {
lock.writeLock().lock();
try {
- // When resizing the cache, we always start with an empty cache. There are two main reasons:
- // 1. Resizing the cache is not a high-frequency operation, and there is no need to fill the data in the old
+ // When resizing the cache, we always start with an empty cache.
+ // Resizing the cache is not a high-frequency operation, and there is no need to fill the data in the old
// cache to the new cache in time when resizing inside.
- // 2. Since the eviction of the caffeine cache is cleared asynchronously, it is possible that after the entry
- // in the old cache is filled in the new cache, the old cache will clear the entry, and the data in the two caches
- // will be inconsistent.
- internalCache.invalidateAll();
+ removeAll(internalCache.asMap().keySet());
Review Comment:
Reverted the change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363626898
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 During read `MarkforCleanUp` should not be called not even once as per the functionality.
Then why we need to call it explicitly again here `invocation.callRealMethod()`.
I am seeing two invocation of `markForCleanUp`
1. `}).when(spyEntry).markForCleanup()` 712 line no
2. `invocation.callRealMethod()` 708 line no
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1362349672
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
Also I found below two test cases flaky which doesn't seems to be related to this change.
- testIndexFileAlreadyExistOnDiskButNotInCache
- testCorrectnessForCacheAndIndexFilesWhenResizeCache
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1365030698
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
> Do you mean this is a mock method and no rename would happen in this case ?
Effectively the functionality/logic of markCleanUp is called one time only ?
Correct.
> // validate cache size based on the file existence`
> if offset file exists validate this
> // validate rsm call should happen if( execution order is remove,read)
> if cache size == 0
> // validate no rsm call should happen if ( execution order is read,remove)
Yes, they are basically similar with what we have now. By injecting mock implementation for `markForCleanUp` is just to make the 2 thread execution more close. In the end, what we have now is to invoke "realMethod", which is what you did above. I'm fine if you think we should validate the rsm call count. But again, they are basically testing the same thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1358349535
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,73 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ val entry = cache.getIndexEntry(rlsMetadata)
+ assertTrue(Files.exists(entry.offsetIndex().file().toPath))
Review Comment:
Yes, you are right.
I have removed the `cache.getIndexEntry` call and just checking that new entry should exist in cache and storage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363708726
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@iit2009060 If I don't call `invocation.callRealMethod()` actual markForCleanup() function will never be called and indexes will not be renamed so how can we test the functionality?
If I remove that `invocation.callRealMethod()` call then markForCleanup() will never be invoked and indexes will never be renamed and hence this test case will fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363534162
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
`testIndexFileAlreadyExistOnDiskButNotInCache` is failing locally sometimes with the same failure mode as `testCorrectnessForCacheAndIndexFilesWhenResizeCache`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1360731330
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@showuon Yes seems correct , as we do not have much control which will get executed we can assert Cache size.
As there may be a case where `remove` and `read` operation has a bug , and nothing happens , but this test case would still pass as file still exists. Adding one more validation when file exists would ensure the correctness of the test case.
```
if (Files.exists(entry.offsetIndex().file().toPath)) {
assertCacheSize(1)
// validate the timestamp of the file created before running the `remove` and `read` operation concurrently with the file recreated after the operation and it should be lesser than the later.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1348641063
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -568,27 +621,26 @@ class RemoteIndexCacheTest {
}
private def verifyFetchIndexInvocation(count: Int,
- indexTypes: Seq[IndexType] =
- Seq(IndexType.OFFSET, IndexType.TIMESTAMP, IndexType.TRANSACTION)): Unit = {
+ indexTypes: Seq[IndexType]): Unit = {
for (indexType <- indexTypes) {
verify(rsm, times(count)).fetchIndex(any(classOf[RemoteLogSegmentMetadata]), ArgumentMatchers.eq(indexType))
}
}
private def createTxIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TransactionIndex = {
- val txnIdxFile = remoteTransactionIndexFile(tpDir, metadata)
+ val txnIdxFile = remoteTransactionIndexFile(new File(tpDir, DIR_NAME), metadata)
txnIdxFile.createNewFile()
new TransactionIndex(metadata.startOffset(), txnIdxFile)
}
private def createTimeIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TimeIndex = {
val maxEntries = (metadata.endOffset() - metadata.startOffset()).asInstanceOf[Int]
- new TimeIndex(remoteTimeIndexFile(tpDir, metadata), metadata.startOffset(), maxEntries * 12)
+ new TimeIndex(remoteTimeIndexFile(new File(tpDir, DIR_NAME), metadata), metadata.startOffset(), maxEntries * 12)
Review Comment:
@jeel2420
There are three functions which is used to create Indexes on tpDir storage.
```
private def createTimeIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TimeIndex
private def createOffsetIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): OffsetIndex
private def createTxIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TransactionIndex
```
tpDir storage in the current unit test act as a data place fetched from remote storage.
The cases where tpDir has been used instead of remoteCacheDir is only when you are creating the spyEntry. But if you see the test cases where it has happened , they are just testing the cache functionality and not mixing it with the remote storage manager functionality.
To support the use cases mentioned above you can do this step
**Suggestion:**
Change the signature of the below method to take Directory path also has an parameter , and refactor the existing test case
```
private def createTimeIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata,dir:File): TimeIndex
private def createOffsetIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata,dir:File): OffsetIndex
private def createTxIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata,dir:File):TransactionIndex
```
Then you can pass the appropriate directory required for your test case.
cc @showuon @divijvaidya
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
@showuon @jeel2420
Should not we use expiredIndxes.offer instead of expiredIndexes.put()
as put will block the operation if queue size is full ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1365030698
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
> Do you mean this is a mock method and no rename would happen in this case ?
Effectively the functionality/logic of markCleanUp is called one time only ?
Correct.
> // validate cache size based on the file existence`
> if offset file exists validate this
> // validate rsm call should happen if( execution order is remove,read)
> if cache size == 0
> // validate no rsm call should happen if ( execution order is read,remove)
Yes, they are basically similar with what we have now. By injecting mock implementation for `markForCleanUp` is just to make the 2 thread execution more close. In the end, what we have now is to invoke "realMethod", which is what you did above. I'm fine if you think we should velidate the rsm call count. But again, they are basically testing the same thing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1773469285
> @jeel2420 https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14483/16/testReport/junit/kafka.log.remote/RemoteIndexCacheTest/Build___JDK_8_and_Scala_2_12___testClearCacheAndIndexFilesWhenResizeCache__/ test is failing in CI. Please check the reason.
I found that it’s a flaky test case. It fails sometimes when I tested locally as well. I will take look into it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364998096
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@showuon
`For (1), the markForCleanUp is an injected method for controlling the invoking order. So there are latches wait/countdown.`
Do you mean this is a mock method and no rename would happen in this case ?
Effectively the functionality/logic of `markCleanUp` is called one time only ?
I was thinking something like this
```
`val latchForTestWait = new CountDownLatch(2)
val removeCache = (() => {
cache.remove(rlsMetadata.remoteLogSegmentId().id())
latchForTestWait.countdown()
}): Runnable
val readCache = (() => {
entry = cache.getIndexEntry(rlsMetadata)
// Signal the CacheRemove to start renaming the files
latchForTestWait.countDown()
}): Runnable
val executor = Executors.newFixedThreadPool(2)
try {
executor.submit(removeCache: Runnable)
executor.submit(readCache: Runnable)
// Wait for signal to complete the test
latchForTestWait.await()
// validate cache size based on the file existence`
if offset file exists validate this
// validate rsm call should happen if( execution order is remove,read)
if cache size == 0
// validate no rsm call should happen if ( execution order is read,remove)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1347002486
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -538,7 +533,75 @@ class RemoteIndexCacheTest {
assertEquals(RemoteIndexCache.DIR_NAME, offsetIndexFile.getParent.getFileName.toString,
s"offsetIndex=$offsetIndexFile is not overwrite under incorrect parent")
// file is corrupted it should fetch from remote storage again
- verifyFetchIndexInvocation(count = 1)
+ verifyFetchIndexInvocation(count = 1, indexTypes = Seq(IndexType.OFFSET))
+ }
+
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ } else {
+ // Subsequent call for markForCleanup method
+ latchForCacheRead.countDown()
+ latchForCacheRemove.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ assertCacheSize(1)
+ val entry = cache.getIndexEntry(rlsMetadata)
+ assertTrue(Files.exists(entry.offsetIndex().file().toPath))
Review Comment:
Please help me understand these lines. My understanding is:
In L598, we want to verify the `cache.getIndexEntry` will be run after `cache.remove` completion. So, that's where the cacheSize 1 comes frome.
If so, then, In L599, why should we `getIndexEntry` again before asserting L600?
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -538,7 +533,75 @@ class RemoteIndexCacheTest {
assertEquals(RemoteIndexCache.DIR_NAME, offsetIndexFile.getParent.getFileName.toString,
s"offsetIndex=$offsetIndexFile is not overwrite under incorrect parent")
// file is corrupted it should fetch from remote storage again
- verifyFetchIndexInvocation(count = 1)
+ verifyFetchIndexInvocation(count = 1, indexTypes = Seq(IndexType.OFFSET))
+ }
+
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ } else {
+ // Subsequent call for markForCleanup method
+ latchForCacheRead.countDown()
+ latchForCacheRemove.countDown()
+ }
Review Comment:
when will this be called?
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -132,7 +133,7 @@ class RemoteIndexCacheTest {
// this call should have invoked fetchOffsetIndex, fetchTimestampIndex once
val resultPosition = cache.lookupOffset(rlsMetadata, offsetPosition1.offset)
assertEquals(offsetPosition1.position, resultPosition)
- verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET, IndexType.TIMESTAMP))
+ verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET))
Review Comment:
I'm not following this change. Do you mean it will fail if we keep `Seq(IndexType.OFFSET, IndexType.TIMESTAMP)`? From my understanding, all 3 index types will be invoked once if cache miss. So, I think we can just call `verifyFetchIndexInvocation(count = 1)` to verify all 3 types.
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -568,27 +631,26 @@ class RemoteIndexCacheTest {
}
private def verifyFetchIndexInvocation(count: Int,
- indexTypes: Seq[IndexType] =
- Seq(IndexType.OFFSET, IndexType.TIMESTAMP, IndexType.TRANSACTION)): Unit = {
+ indexTypes: Seq[IndexType]): Unit = {
Review Comment:
I don't think this change is necessary. When fetchIndex, we'll always fetch all 3 types of indexes. So why can't we verify all of them by default?
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -281,14 +282,8 @@ class RemoteIndexCacheTest {
// no expired entries yet
assertEquals(0, cache.expiredIndexes.size, "expiredIndex queue should be zero at start of test")
- // invalidate the cache. it should async mark the entry for removal
- cache.internalCache.invalidate(internalIndexKey)
-
- // wait until entry is marked for deletion
- TestUtils.waitUntilTrue(() => cacheEntry.isMarkedForCleanup,
- "Failed to mark cache entry for cleanup after invalidation")
- TestUtils.waitUntilTrue(() => cacheEntry.isCleanStarted,
- "Failed to cleanup cache entry after invalidation")
+ // call remove function to mark the entry for removal
+ cache.remove(internalIndexKey)
// first it will be marked for cleanup, second time markForCleanup is called when cleanup() is called
verify(cacheEntry, times(2)).markForCleanup()
Review Comment:
In L291, we want to verify `cleanup` is invoked, but after my suggestion to remove `cleanup` in `internalCache.asMap().computeIfPresent`, now, we need to add this before verify it.
```
TestUtils.waitUntilTrue(() => cacheEntry.isCleanStarted,
"Failed to cleanup cache entry after invalidation")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1358347800
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -157,9 +154,9 @@ private Cache<Uuid, Entry> initEmptyCache(long maxSize) {
.weigher((Uuid key, Entry entry) -> {
return (int) entry.entrySizeBytes;
})
- // removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
+ // evictionListener is invoked when either the entry is invalidated (means manual removal by the caller) or
// evicted (means removal due to the policy)
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363504542
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 `markForCleanUp` should be called only one time. We need to test the behaviour when there are concurrent read/remove happens on the cache for the same entry.
In the test we just need to assert the way @showuon suggested
// So, maybe we verify with this:
if (Files.exists(entry.offsetIndex().file().toPath)) {
assertCacheSize(1)
} else {
assertCacheSize(0)
}
Calling '`markForCleanUp`' twice will always result in cacheSize 0 eventually.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363633722
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ entry = cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ if (getIndexFileFromRemoteCacheDir(cache, LogFileUtils.INDEX_FILE_SUFFIX).isPresent) {
+ assertCacheSize(1)
+ } else {
+ assertCacheSize(0)
+ }
Review Comment:
It's worth adding some comments here to explain why we need to verify 2 different cases. Ex:
`We can't determine read thread or remove thread will go first, but we want to make sure when cache existed, the cache file should exist, and if cache is non-existed, the cache file should not exist.`
Something like it. You can make it more clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1352268556
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -169,7 +169,16 @@ public Cache<Uuid, Entry> internalCache() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ v.cleanup();
Review Comment:
Done
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -178,7 +187,16 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364855206
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 There are two times when markForCleanUp is called.
1. `remove` function which we are calling in `removeCache Runnable`.
2. One at `invocation.callRealMethod() 708 line no`
Even i tried running your test case locally it always assert with cacheSize 0 , as it is eventually getting deleted.
IMO we should read and remove concurrently in the separate thread and validate the cacheSize based on the order of execution. We should not need to call explicitly for the scenario.
@showuon WDYT ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1366846188
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -151,9 +151,8 @@ private Cache<Uuid, Entry> initEmptyCache(long maxSize) {
.weigher((Uuid key, Entry entry) -> {
return (int) entry.entrySizeBytes;
})
- // removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
- // evicted (means removal due to the policy)
- .removalListener((Uuid key, Entry entry, RemovalCause cause) -> {
+ // evictionListener is invoked when RemovalCause.wasEvicted() is true
Review Comment:
Done
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -187,7 +186,18 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ if (!expiredIndexes.offer(v)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", v, k);
+ }
+ } catch (IOException e) {
+ throw new KafkaException(e);
+ }
+ // Returning null to remove the key from the cache
+ return null;
+ });
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1379849170
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
+ }
+ } catch (IOException e) {
Review Comment:
(please correct me if I am wrong) That is not the behaviour for TS in Kafka since we don't want to mark local directory as offline if there is something wrong with functioning of TS. As an example, on IOException on copyLogSegment, we will eat up the exception at https://github.com/apache/kafka/blob/244abc11afce37e8b11c29b5427ab5042920cd55/core/src/main/java/kafka/log/remote/RemoteLogManager.java#L700
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1380563100
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
Review Comment:
I am not sure reliance on start-up clean-up is satisfying enough because there is no guarantee the file system hasn't be saturated when the broker restarts and that any clean-up operation will succeed. Also, it does not address the risk (albeit minor) of the unbounded queue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364955572
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
> There are two times when markForCleanUp is called.
> 1. remove function which we are calling in removeCache Runnable.
> 2. One at invocation.callRealMethod() 708 line no
You're right, but they are "different" `markForCleanUp`.
For (1), the `markForCleanUp` is an injected method for controlling the invoking order. So there are latches wait/countdown.
For (2), it's the real `markForCleanUp` method to rename the cache files.
The goal is to simulate the race condition happened in [KAFKA-15481](https://issues.apache.org/jira/browse/KAFKA-15481).
> Even i tried running your test case locally it always assert with cacheSize 0 , as it is eventually getting deleted.
Yes, I think so. But in some cases, there could be 1 if getEntry goes after. The thread management are all decided by OS, we can't assure which one will go first, right?
I think the goal of this test is to make sure the issue in [KAFKA-15481](https://issues.apache.org/jira/browse/KAFKA-15481) will not happen again. That's why I added this [comment](https://github.com/apache/kafka/pull/14483#discussion_r1364844466).
> IMO we should read and remove concurrently in the separate thread and validate the cacheSize based on the order of execution.
I'm not following you here. What we're doing in this test is to `read and remove concurrently in the separate thread`. About `validate the cacheSize based on the order of execution`, since we can't assure which thread will be executed first, we can't do this, right? If we can decide the execution order, then it means they are not running _concurrently_, is that right?
> We should not need to call explicitly for the scenario.
Maybe you can show us if it were you, what test you'll create. Some pseudo code are enough. Thank you.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363633722
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ entry = cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ if (getIndexFileFromRemoteCacheDir(cache, LogFileUtils.INDEX_FILE_SUFFIX).isPresent) {
+ assertCacheSize(1)
+ } else {
+ assertCacheSize(0)
+ }
Review Comment:
It's worth adding some comments here to explain why we need to verify 2 different cases. Ex:
`We can't determine read thread or remove thread will go first, but we have to want to make sure when cache existed, the cache file should exist, too, and if cache is non-existed, the cache file should not exist.`
Something like it. You can make it more clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1366804550
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -151,9 +151,8 @@ private Cache<Uuid, Entry> initEmptyCache(long maxSize) {
.weigher((Uuid key, Entry entry) -> {
return (int) entry.entrySizeBytes;
})
- // removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
- // evicted (means removal due to the policy)
- .removalListener((Uuid key, Entry entry, RemovalCause cause) -> {
+ // evictionListener is invoked when RemovalCause.wasEvicted() is true
Review Comment:
Please update the comment here to explicitly specify whether it would be executed sync or async with cache eviction. You can say:
This listener is invoked each time an entry is being automatically removed due to eviction. The cache with invoke this listener during the atomic operation to remove the entry (refer: https://github.com/ben-manes/caffeine/wiki/Removal), hence, care must be taken to ensure that this operation is not expensive.
Note that this listener is not invoked when RemovalCause from cache is EXPLICIT or REPLACED (e.g. on Cache.invalidate(), Cache.put() etc.) For a complete list see: https://github.com/ben-manes/caffeine/blob/0cef55168986e3816314e7fdba64cb0b996dd3cc/caffeine/src/main/java/com/github/benmanes/caffeine/cache/RemovalCause.java#L23. Hence, any operation required after removal from cache must be performed manually for these scenarios.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1366804550
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -151,9 +151,8 @@ private Cache<Uuid, Entry> initEmptyCache(long maxSize) {
.weigher((Uuid key, Entry entry) -> {
return (int) entry.entrySizeBytes;
})
- // removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
- // evicted (means removal due to the policy)
- .removalListener((Uuid key, Entry entry, RemovalCause cause) -> {
+ // evictionListener is invoked when RemovalCause.wasEvicted() is true
Review Comment:
Please update the comment here to explicitly specify whether it would be executed sync or async with cache eviction. You can say:
This listener is invoked each time an entry is being automatically removed due to eviction. The cache with invoke this listener during the atomic operation to remove the entry (refer: https://github.com/ben-manes/caffeine/wiki/Removal), hence, care must be taken to ensure that this operation is not expensive.
Note that this listener is not invoked when RemovalCause from cache is EXPLICIT or REPLACED (e.g. on Cache.invalidate(), Cache.put() etc.) For a complete list see: https://github.com/ben-manes/caffeine/blob/e12bb5cf8a6d1b1318835ad1b63901cd338de3dc/caffeine/src/main/java/com/github/benmanes/caffeine/cache/RemovalCause.java#L23. Hence, any operation required after removal from cache must be performed manually for these scenarios.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1361468155
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
Another PR with `MINOR` prefixed like this: https://github.com/apache/kafka/pull/14559
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364855206
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 There are two times when markForCleanUp is called.
1. Throw `remove` function which we are calling in `removeCache Runnable`.
2. One at `invocation.callRealMethod() 708 line no`
Even i tried running your test case locally it always assert with cacheSize 0 , as it is eventually getting deleted.
IMO we should read and remove concurrently in the separate thread and validate the cacheSize based on the order of execution. We should not need to call explicitly for the scenario.
@showuon WDYT ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364998096
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@showuon
`For (1), the markForCleanUp is an injected method for controlling the invoking order. So there are latches wait/countdown.`
Do you mean this is a mock method and no rename would happen in this case ?
Effectively the functionality/logic of `markCleanUp` is called one time only ?
I was thinking something like this
`val latchForTestWait = new CountDownLatch(2)
val removeCache = (() => {
cache.remove(rlsMetadata.remoteLogSegmentId().id())
latchForTestWait.countdown()
}): Runnable
val readCache = (() => {
entry = cache.getIndexEntry(rlsMetadata)
// Signal the CacheRemove to start renaming the files
latchForTestWait.countDown()
}): Runnable
val executor = Executors.newFixedThreadPool(2)
try {
executor.submit(removeCache: Runnable)
executor.submit(readCache: Runnable)
// Wait for signal to complete the test
latchForTestWait.await()
// validate cache size `
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1365036479
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@showuon Yes correct , it is testing the same thing. I am also fine. But from readability perspective the one I propose is simpler to understand and does not require any future change if markForCleanUp function changed. I left @jeel2420 to make a decision here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1348624218
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -568,27 +621,26 @@ class RemoteIndexCacheTest {
}
private def verifyFetchIndexInvocation(count: Int,
- indexTypes: Seq[IndexType] =
- Seq(IndexType.OFFSET, IndexType.TIMESTAMP, IndexType.TRANSACTION)): Unit = {
+ indexTypes: Seq[IndexType]): Unit = {
for (indexType <- indexTypes) {
verify(rsm, times(count)).fetchIndex(any(classOf[RemoteLogSegmentMetadata]), ArgumentMatchers.eq(indexType))
}
}
private def createTxIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TransactionIndex = {
- val txnIdxFile = remoteTransactionIndexFile(tpDir, metadata)
+ val txnIdxFile = remoteTransactionIndexFile(new File(tpDir, DIR_NAME), metadata)
txnIdxFile.createNewFile()
new TransactionIndex(metadata.startOffset(), txnIdxFile)
}
private def createTimeIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TimeIndex = {
val maxEntries = (metadata.endOffset() - metadata.startOffset()).asInstanceOf[Int]
- new TimeIndex(remoteTimeIndexFile(tpDir, metadata), metadata.startOffset(), maxEntries * 12)
+ new TimeIndex(remoteTimeIndexFile(new File(tpDir, DIR_NAME), metadata), metadata.startOffset(), maxEntries * 12)
Review Comment:
The reason behind storing the files in cacheDir is below.
In unit test case `testConcurrentRemoveReadForCache` which reproduce the concurrency bug, I am first creating the index files and creating a spyEntry and adding it in the Cache. After calling the remove function for that spyEntry and before calling the markForCleanup(), when I am calling getIndexEntry without the above change (i.e creating the files in tpDir), in loadIndexFile() it is trying to fetch the index from remote again which I don't think is right as spyEntry is not yet renamed and it should be picked from the file.
Also, once the markForCleanup() called, the index files are getting renamed into tpDir only if we create the index files in tpDir and actual index files are getting stored in cacheDir so that way it is not possible to reproduce this bug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1347046103
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -132,7 +133,7 @@ class RemoteIndexCacheTest {
// this call should have invoked fetchOffsetIndex, fetchTimestampIndex once
val resultPosition = cache.lookupOffset(rlsMetadata, offsetPosition1.offset)
assertEquals(offsetPosition1.position, resultPosition)
- verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET, IndexType.TIMESTAMP))
+ verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET))
Review Comment:
In the fetchIndex mock (in setup()), irrespective of the IndexType passed in the parameter, we are creating all the index files in the cacheDir and due to that, during cacheMiss, fetchIndex is getting invoked during OFFSET index fetch only and for other two index types, fetchIndex is not getting called as the index is getting fetched from the disk only.
These test cases were passing earlier, as before we were creating those index files in tmpDir, not in the actual dir used for RemoteIndexCache, and after I fixed that issue, test cases started to fail, so I had to make this change and similar changes in other test cases as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1347129743
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -281,14 +282,8 @@ class RemoteIndexCacheTest {
// no expired entries yet
assertEquals(0, cache.expiredIndexes.size, "expiredIndex queue should be zero at start of test")
- // invalidate the cache. it should async mark the entry for removal
- cache.internalCache.invalidate(internalIndexKey)
-
- // wait until entry is marked for deletion
- TestUtils.waitUntilTrue(() => cacheEntry.isMarkedForCleanup,
- "Failed to mark cache entry for cleanup after invalidation")
- TestUtils.waitUntilTrue(() => cacheEntry.isCleanStarted,
- "Failed to cleanup cache entry after invalidation")
+ // call remove function to mark the entry for removal
+ cache.remove(internalIndexKey)
// first it will be marked for cleanup, second time markForCleanup is called when cleanup() is called
verify(cacheEntry, times(2)).markForCleanup()
Review Comment:
I removed `cleanup` call in `internalCache.asMap().computeIfPresent` and added wait as well but cleanup is not getting called.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1348242651
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -568,27 +621,26 @@ class RemoteIndexCacheTest {
}
private def verifyFetchIndexInvocation(count: Int,
- indexTypes: Seq[IndexType] =
- Seq(IndexType.OFFSET, IndexType.TIMESTAMP, IndexType.TRANSACTION)): Unit = {
+ indexTypes: Seq[IndexType]): Unit = {
for (indexType <- indexTypes) {
verify(rsm, times(count)).fetchIndex(any(classOf[RemoteLogSegmentMetadata]), ArgumentMatchers.eq(indexType))
}
}
private def createTxIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TransactionIndex = {
- val txnIdxFile = remoteTransactionIndexFile(tpDir, metadata)
+ val txnIdxFile = remoteTransactionIndexFile(new File(tpDir, DIR_NAME), metadata)
txnIdxFile.createNewFile()
new TransactionIndex(metadata.startOffset(), txnIdxFile)
}
private def createTimeIndexForSegmentMetadata(metadata: RemoteLogSegmentMetadata): TimeIndex = {
val maxEntries = (metadata.endOffset() - metadata.startOffset()).asInstanceOf[Int]
- new TimeIndex(remoteTimeIndexFile(tpDir, metadata), metadata.startOffset(), maxEntries * 12)
+ new TimeIndex(remoteTimeIndexFile(new File(tpDir, DIR_NAME), metadata), metadata.startOffset(), maxEntries * 12)
Review Comment:
Remote Storage Manager mock should always creates file in tpDir directory and not under cacheDir.
It is the job of the remoteindexcache to copy it from tpDir and creates under remote-index-cache dir.
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -132,7 +133,7 @@ class RemoteIndexCacheTest {
// this call should have invoked fetchOffsetIndex, fetchTimestampIndex once
val resultPosition = cache.lookupOffset(rlsMetadata, offsetPosition1.offset)
assertEquals(offsetPosition1.position, resultPosition)
- verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET, IndexType.TIMESTAMP))
+ verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET))
Review Comment:
@jeel2420 The fetchindex mock creates each segment in the tpDir directory and not in the cacheDir. Files will be created in the cacheDir once we make a call to getIndexEntry which copies from tpDir to cacheDir.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "kamalcph (via GitHub)" <gi...@apache.org>.
kamalcph commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1349878452
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -134,7 +134,7 @@ public RemoteIndexCache(int maxSize, RemoteStorageManager remoteStorageManager,
.maximumSize(maxSize)
// removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
Review Comment:
Can we update the comment to refer the `evictionListener` instead of `removeListener`?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -169,7 +169,16 @@ public Cache<Uuid, Entry> internalCache() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ v.cleanup();
Review Comment:
Also, we have to add that entry to the `expiredIndexes` so that cleanup thread will be invoked on the removed entry.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1356780374
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -132,7 +133,7 @@ class RemoteIndexCacheTest {
// this call should have invoked fetchOffsetIndex, fetchTimestampIndex once
val resultPosition = cache.lookupOffset(rlsMetadata, offsetPosition1.offset)
assertEquals(offsetPosition1.position, resultPosition)
- verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET, IndexType.TIMESTAMP))
+ verifyFetchIndexInvocation(count = 1, Seq(IndexType.OFFSET))
Review Comment:
@iit2009060 Got it. I have reverted these changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1362346109
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
Updated test case.
@iit2009060 I test the test case without the fix and test case is failing as expected so I haven't added timestamp. Can you please tell me in which particular case the test case could pass with the older code (without fix).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1365704660
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@iit2009060 As current test case is able to reproduce the case mentioned in the jira [KAFKA-15481](https://issues.apache.org/jira/browse/KAFKA-15481) I think we should be fine.
The only reason to have `markForCleanUp` mock is to have control over the 2 thread execution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364844466
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,89 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ entry = cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
Review Comment:
@jeel2420 , sorry, I had another look and found we should also verify these 2 threads has no exception thrown. In the issue description, without this fix, there will be IOException thrown. So, we should verify there's no exception using the returned `future` from `executor.submit`. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363459855
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 I am not seeing `testIndexFileAlreadyExistOnDiskButNotInCache` this as a failure in integration test.
@hudeqi testCorrectnessForCacheAndIndexFilesWhenResizeCache can you help here , I see we already handled NoFileException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363492981
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
invocation.callRealMethod() is called to call the markForCleanup() after read is called and before we start asserting to make sure indexes get renamed before we assert the results.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363533239
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@iit2009060 Read `markForCleanUp()` is not getting called twice. Please see the mock, inside that during the first execution of `markForCleanUp()` I am calling the actual markForCleanup() function (i.e Index are getting renamed) but for subsequent calls, mock is doing nothing so we actual markForCleanup() function to rename the indexes is getting called once only and it is as expected.
I have verified this behaviour as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364955572
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
> There are two times when markForCleanUp is called.
> 1. remove function which we are calling in removeCache Runnable.
> 2. One at invocation.callRealMethod() 708 line no
You're right, but they are "different" `markForCleanUp`.
For (1), the `markForCleanUp` is an injected method for controlling the invoking order.
For (2), it's the real `markForCleanUp` method to rename the cache files.
The goal is to simulate the race condition happened in [KAFKA-15481](https://issues.apache.org/jira/browse/KAFKA-15481).
> Even i tried running your test case locally it always assert with cacheSize 0 , as it is eventually getting deleted.
Yes, I think so. But in some cases, there could be 1 if getEntry goes after. The thread management are all decided by OS, we can't assure which one will go first, right?
I think the goal of this test is to make sure the issue in [KAFKA-15481](https://issues.apache.org/jira/browse/KAFKA-15481) will not happen again. That's why I added this [comment](https://github.com/apache/kafka/pull/14483#discussion_r1364844466).
> IMO we should read and remove concurrently in the separate thread and validate the cacheSize based on the order of execution.
I'm not following you here. What we're doing in this test is to `read and remove concurrently in the separate thread`. About `validate the cacheSize based on the order of execution`, since we can't assure which thread will be executed first, we can't do this, right? If we can decide the execution order, then it means they are not running _concurrently_, is that right?
> We should not need to call explicitly for the scenario.
Maybe you can show us if it were you, what test you'll create. Some pseudo code are enough. Thank you.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1775128178
Created a JIRA for the flaky test https://issues.apache.org/jira/browse/KAFKA-15671
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1773437487
@jeel2420 https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14483/16/testReport/junit/kafka.log.remote/RemoteIndexCacheTest/Build___JDK_8_and_Scala_2_12___testClearCacheAndIndexFilesWhenResizeCache__/ test is failing in CI. Please check the reason.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363446715
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 Why we are invoking this again ` invocation.callRealMethod()`
It is already called in
`712 line when(spyEntry).markForCleanup()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364116455
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 Is it not called in the line below ?
`}).when(spyEntry).markForCleanup() 712 line no`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1387607498
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
+ }
+ } catch (IOException e) {
Review Comment:
Good point @Hangleton ! But I didn't see the JIRA created (maybe I missed). Could you let me know the JIRA ticket number? Or should I create one for you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1365700635
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,89 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ entry = cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
Review Comment:
@showuon Nice catch. Calling .get() on both task future objects so if there will be any error test will fail with that exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya merged PR #14483:
URL: https://github.com/apache/kafka/pull/14483
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1379841108
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
Review Comment:
Correct. It will be leaked with a ".delete" suffix. It should be cleared at next process restart at https://github.com/apache/kafka/blob/244abc11afce37e8b11c29b5427ab5042920cd55/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java#L265
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1380576649
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
+ }
+ } catch (IOException e) {
Review Comment:
The cache stores its files [in the first log directory](https://github.com/apache/kafka/blob/4d8efa94cbe6fd74d23dae2608058db52521bb2c/core/src/main/scala/kafka/server/KafkaServer.scala#L619) defined in server properties. Because I/O exceptions captured here originate from a log directory, why shouldn't the pattern which applies to log directory failures in Kafka not be valid here?
Regarding the I/O exceptions surfaced from `copyLogSegment` - are we talking about those which occurs from within a plugin? If that is the case, then they are of a different nature since not applying to log directories, and by design the plugin implementation needs to handle those while the remote log manager provides a retry mechanism via the periodic scheduling of the RLM tasks. The idea is that unlike file system I/O errors on a log directory which are directly putting local data integrity at risk, transient I/O errors are common for clients transferring data to or from external services e.g. a public cloud storage for which availability recovers on its own and given that durability is not compromised as long as consistency of metadata updates is guaranteed. So the semantic of the I/O exception, the corresponding failure modes, and their implications, are not the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1347075039
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -538,7 +533,75 @@ class RemoteIndexCacheTest {
assertEquals(RemoteIndexCache.DIR_NAME, offsetIndexFile.getParent.getFileName.toString,
s"offsetIndex=$offsetIndexFile is not overwrite under incorrect parent")
// file is corrupted it should fetch from remote storage again
- verifyFetchIndexInvocation(count = 1)
+ verifyFetchIndexInvocation(count = 1, indexTypes = Seq(IndexType.OFFSET))
+ }
+
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ } else {
+ // Subsequent call for markForCleanup method
+ latchForCacheRead.countDown()
+ latchForCacheRemove.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ assertCacheSize(1)
+ val entry = cache.getIndexEntry(rlsMetadata)
+ assertTrue(Files.exists(entry.offsetIndex().file().toPath))
Review Comment:
Here we want to prove that the entry is present in the cache so when we call getIndexEntry we should get the entry and index files should be present in the cacheDir.
This reproduce the actual issue by replicating the scenario where we can have inconsistency where key is present in the cache but reference files are already renamed by that time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1347129743
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -281,14 +282,8 @@ class RemoteIndexCacheTest {
// no expired entries yet
assertEquals(0, cache.expiredIndexes.size, "expiredIndex queue should be zero at start of test")
- // invalidate the cache. it should async mark the entry for removal
- cache.internalCache.invalidate(internalIndexKey)
-
- // wait until entry is marked for deletion
- TestUtils.waitUntilTrue(() => cacheEntry.isMarkedForCleanup,
- "Failed to mark cache entry for cleanup after invalidation")
- TestUtils.waitUntilTrue(() => cacheEntry.isCleanStarted,
- "Failed to cleanup cache entry after invalidation")
+ // call remove function to mark the entry for removal
+ cache.remove(internalIndexKey)
// first it will be marked for cleanup, second time markForCleanup is called when cleanup() is called
verify(cacheEntry, times(2)).markForCleanup()
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1762593153
@jeel2420 , there's a test failed due to this change, please take a look.
https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14483/10/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1361468155
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
Another PR with `MINOR` prefixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364477109
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@iit2009060 I am not quite sure on that.
This mock function is getting called two times. First during the remove call and during that time mock calls the read method to actually rename the index files and it is getting called at the end as well after testWait latch is countdown and during that time mock is doing nothing.
So I don’t think there is any issue here as actual markForCleanup() method is called only once and indexes are getting renamed before we perform the assertions.
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@iit2009060 I am not quite sure on that.
This mock function is getting called two times. First during the remove call and during that time mock calls the read method to actually rename the index files and it is getting called at the end as well after testWait latch is countdown and during that time mock is doing nothing.
So I don’t think there is any issue here as actual markForCleanup() method is called only once and indexes are getting renamed before we perform the assertions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1380576649
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
+ log.error("Error while inserting entry {} for key {} into the cleaner queue because queue is full.", entry, key);
+ }
+ } catch (IOException e) {
Review Comment:
The cache stores its files [in the first log directory](https://github.com/apache/kafka/blob/4d8efa94cbe6fd74d23dae2608058db52521bb2c/core/src/main/scala/kafka/server/KafkaServer.scala#L619) defined in server properties. Because I/O exceptions captured here originate from a log directory, why shouldn't the pattern which applies to log directory failures in Kafka not be valid here?
Regarding the I/O exceptions surfaced from `copyLogSegment` - are we talking about those which occurs from within a plugin? If that is the case, then they are of a different nature since not applying to log directories, and by design the plugin implementation needs to handle those while the remote log manager provides a retry mechanism via the periodic scheduling of the RLM tasks. The idea is that unlike file system I/O errors on a log directory which are directly putting local data integrity at risk, transient I/O errors are common for clients transferring data to or from external services e.g. a public cloud storage. So the semantic of the I/O exception, the corresponding failure modes, and their implications, are not the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1382984874
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -196,12 +197,27 @@ public void remove(Uuid key) {
public void removeAll(Collection<Uuid> keys) {
lock.readLock().lock();
try {
- internalCache.invalidateAll(keys);
+ keys.forEach(key -> internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ enqueueEntryForCleanup(v, k);
+ // Returning null to remove the key from the cache
+ return null;
+ }));
} finally {
lock.readLock().unlock();
}
}
+ private void enqueueEntryForCleanup(Entry entry, Uuid key) {
+ try {
+ entry.markForCleanup();
+ if (!expiredIndexes.offer(entry)) {
Review Comment:
Fair enough. Would you like to create an improvement JIRA to discuss the alternative solutions to this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1359202160
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
Use `offer`. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1359185319
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,75 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
Why we are calling actual rename of the file in callRealMethod ?
Correct me if my understanding is wrong here ?
Two threads are defined
1. removalCache - executing remove function of the cache.
2. ReadCache - for reading data from the cache when spy.markforEntryCleanUp is executed.
=====
Operations
1. removeCache triggered
2. spyEntry.markForCleanup when executed ( the files are already renamed to .delete)
3. readCache executed and finished because no lock is pending on the remove operation
4. It creates new file in the entry again(fetched from remote storage, rather than existed in the cache)(We should validate the number of calls to rsm here )
5.After latchfoCacheRemove.await() ,why we are explicitlly calling markCleanup again ?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
@showuon I was just thinking around it , if this fails , it will create a dump of files with delete suffix entry which never gets deleted from the disk ? Is the behaviour ok ?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
@showuon @kamalcph @jeel2420 I think we are entering into a deadlock state here
During remove we try to take a readLock but in markForCleanUp we try to take a write lock , Will it not result a deadlock state ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1357702660
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -137,13 +137,10 @@ public RemoteIndexCache(long maxSize, RemoteStorageManager remoteStorageManager,
public void resizeCacheSize(long remoteLogIndexFileCacheSize) {
lock.writeLock().lock();
try {
- // When resizing the cache, we always start with an empty cache. There are two main reasons:
- // 1. Resizing the cache is not a high-frequency operation, and there is no need to fill the data in the old
+ // When resizing the cache, we always start with an empty cache.
+ // Resizing the cache is not a high-frequency operation, and there is no need to fill the data in the old
// cache to the new cache in time when resizing inside.
- // 2. Since the eviction of the caffeine cache is cleared asynchronously, it is possible that after the entry
- // in the old cache is filled in the new cache, the old cache will clear the entry, and the data in the two caches
- // will be inconsistent.
- internalCache.invalidateAll();
+ removeAll(internalCache.asMap().keySet());
Review Comment:
OK, we can keep this change, there will be update at this: https://github.com/apache/kafka/pull/14511 . cc @hudeqi
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -592,16 +593,73 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ doAnswer((invocation: InvocationOnMock) => {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ val entry = cache.getIndexEntry(rlsMetadata)
+ assertTrue(Files.exists(entry.offsetIndex().file().toPath))
Review Comment:
We need to verify something between L644 and L645, that is, after 2 threads completed, what state we're expecting?
My understanding is,`spyEntry.markForCleanup` will get invoked before `cache.getIndexEntry` for the same entry, that means, `entry update for getIndexEntry` needs to wait until `entry remove` completed, since they are both atomic operation. In this case, after the L644, the new entry should existed in cache and storage. So, we don't actually need L645 to `getIndexEntry`. Is my understanding correct?
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -157,9 +154,9 @@ private Cache<Uuid, Entry> initEmptyCache(long maxSize) {
.weigher((Uuid key, Entry entry) -> {
return (int) entry.entrySizeBytes;
})
- // removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
+ // evictionListener is invoked when either the entry is invalidated (means manual removal by the caller) or
// evicted (means removal due to the policy)
Review Comment:
evictionListener will be invoked when evicted, as [the doc](https://github.com/ben-manes/caffeine/wiki/Removal) said:
> When the operation must be performed synchronously with eviction, use Caffeine.evictionListener(RemovalListener) instead. This listener will only be notified when RemovalCause.wasEvicted() is true. For an explicit removal, Cache.asMap() offers compute methods that are performed atomically.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1352269114
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -134,7 +134,7 @@ public RemoteIndexCache(int maxSize, RemoteStorageManager remoteStorageManager,
.maximumSize(maxSize)
// removeListener is invoked when either the entry is invalidated (means manual removal by the caller) or
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363708726
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@iit2009060 If I don't call `invocation.callRealMethod()` actual markForCleanup() function will never be called and indexes will not be renamed so how can we test the functionality?
If I remove that `invocation.callRealMethod()` call then markForCleanup() will never be invoked and indexes will never be renamed and hence this test case will fail.
Am I missing something here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364116455
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@jeel2420 Is it not called in the 1st line I mentioned
`}).when(spyEntry).markForCleanup() 712 line no`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1363633722
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
+ // Signal TestWait to unblock itself so that test can be completed
+ latchForTestWait.countDown()
+ }
+ }).when(spyEntry).markForCleanup()
+
+ val removeCache = (() => {
+ cache.remove(rlsMetadata.remoteLogSegmentId().id())
+ }): Runnable
+
+ val readCache = (() => {
+ // Wait for signal to start CacheRead
+ latchForCacheRead.await()
+ entry = cache.getIndexEntry(rlsMetadata)
+ // Signal the CacheRemove to start renaming the files
+ latchForCacheRemove.countDown()
+ }): Runnable
+
+ val executor = Executors.newFixedThreadPool(2)
+ try {
+ executor.submit(removeCache: Runnable)
+ executor.submit(readCache: Runnable)
+
+ // Wait for signal to complete the test
+ latchForTestWait.await()
+ if (getIndexFileFromRemoteCacheDir(cache, LogFileUtils.INDEX_FILE_SUFFIX).isPresent) {
+ assertCacheSize(1)
+ } else {
+ assertCacheSize(0)
+ }
Review Comment:
It's worth adding some comments here to explain why we need to verify 2 different cases. Ex:
`We can't determine read thread or remove thread will go first, but we want to make sure when cache existed, the cache file should exist, too, and if cache is non-existed, the cache file should not exist.`
Something like it. You can make it more clear.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1364998096
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -672,16 +673,84 @@ class RemoteIndexCacheTest {
verifyFetchIndexInvocation(count = 1)
}
+ @Test
+ def testConcurrentRemoveReadForCache(): Unit = {
+ // Create a spy Cache Entry
+ val rlsMetadata = new RemoteLogSegmentMetadata(RemoteLogSegmentId.generateNew(idPartition), baseOffset, lastOffset,
+ time.milliseconds(), brokerId, time.milliseconds(), segmentSize, Collections.singletonMap(0, 0L))
+
+ val timeIndex = spy(createTimeIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val txIndex = spy(createTxIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+ val offsetIndex = spy(createOffsetIndexForSegmentMetadata(rlsMetadata, new File(tpDir, DIR_NAME)))
+
+ val spyEntry = spy(new RemoteIndexCache.Entry(offsetIndex, timeIndex, txIndex))
+ cache.internalCache.put(rlsMetadata.remoteLogSegmentId().id(), spyEntry)
+
+ assertCacheSize(1)
+
+ var entry: RemoteIndexCache.Entry = null
+
+ val latchForCacheRead = new CountDownLatch(1)
+ val latchForCacheRemove = new CountDownLatch(1)
+ val latchForTestWait = new CountDownLatch(1)
+
+ var markForCleanupCallCount = 0
+
+ doAnswer((invocation: InvocationOnMock) => {
+ markForCleanupCallCount += 1
+
+ if (markForCleanupCallCount == 1) {
+ // Signal the CacheRead to unblock itself
+ latchForCacheRead.countDown()
+ // Wait for signal to start renaming the files
+ latchForCacheRemove.await()
+ // Calling the markForCleanup() actual method to start renaming the files
+ invocation.callRealMethod()
Review Comment:
@showuon
`For (1), the markForCleanUp is an injected method for controlling the invoking order. So there are latches wait/countdown.`
Do you mean this is a mock method and no rename would happen in this case ?
Effectively the functionality/logic of `markCleanUp` is called one time only ?
I was thinking something like this
```
`val latchForTestWait = new CountDownLatch(2)
val removeCache = (() => {
cache.remove(rlsMetadata.remoteLogSegmentId().id())
latchForTestWait.countdown()
}): Runnable
val readCache = (() => {
entry = cache.getIndexEntry(rlsMetadata)
// Signal the CacheRemove to start renaming the files
latchForTestWait.countDown()
}): Runnable
val executor = Executors.newFixedThreadPool(2)
try {
executor.submit(removeCache: Runnable)
executor.submit(readCache: Runnable)
// Wait for signal to complete the test
latchForTestWait.await()
// validate cache size based on the file existence`
if offset file exists validate this
// validate rsm call should happen if( execution order is remove,read)
if cache size == 0
// validate no rsm call should happen if ( execution order is read,remove)
```
In the test case mentioned in the jira [KAFKA-15481](https://issues.apache.org/jira/browse/KAFKA-15481)
the execution order is remove,read and the overall result is cache size 0 which is wrong because of timegap between removal and renaming the files. Here we are validating the same with rsm call count. If they are atomic rsm execution should happen and files should be restored.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1771965660
> @iit2009060 , do you have any comments to this PR?
@showuon No , I am good . Thanks @jeel2420 for addressing the review comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on PR #14483:
URL: https://github.com/apache/kafka/pull/14483#issuecomment-1771958677
@iit2009060 , do you have any comments to this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "iit2009060 (via GitHub)" <gi...@apache.org>.
iit2009060 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1360731604
##########
storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java:
##########
@@ -193,7 +192,16 @@ public File cacheDir() {
public void remove(Uuid key) {
lock.readLock().lock();
try {
- internalCache.invalidate(key);
+ internalCache.asMap().computeIfPresent(key, (k, v) -> {
+ try {
+ v.markForCleanup();
+ expiredIndexes.put(v);
Review Comment:
Yah sure @showuon Let me do it. Should I create a separate ticket or another PR attached with the same ticket ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15481: Fix concurrency bug in RemoteIndexCache [kafka]
Posted by "jeel2420 (via GitHub)" <gi...@apache.org>.
jeel2420 commented on code in PR #14483:
URL: https://github.com/apache/kafka/pull/14483#discussion_r1347056681
##########
core/src/test/scala/unit/kafka/log/remote/RemoteIndexCacheTest.scala:
##########
@@ -568,27 +631,26 @@ class RemoteIndexCacheTest {
}
private def verifyFetchIndexInvocation(count: Int,
- indexTypes: Seq[IndexType] =
- Seq(IndexType.OFFSET, IndexType.TIMESTAMP, IndexType.TRANSACTION)): Unit = {
+ indexTypes: Seq[IndexType]): Unit = {
Review Comment:
Addressed this comment here: https://github.com/apache/kafka/pull/14483#discussion_r1347046103
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org