java.lang.InterruptedException when tring to read events from remote storage

[catalinmer]

Hi,

After running it without problems for several weeks (I guess mostly we put events into remote storage, there wasn't much reading until now) we started to have the following errors in the logs.
Looks like it is trying to read some events from remote storage but it fails to do so, below it is the error:

ERROR Error occurred while reading the remote data for wallet-events-3 (kafka.log.remote.RemoteLogReader)
org.apache.kafka.common.KafkaException: org.apache.kafka.server.log.remote.storage.RemoteStorageException: java.lang.RuntimeException: java.lang.InterruptedException
	at org.apache.kafka.storage.internals.log.RemoteIndexCache.lambda$createCacheEntry$13(RemoteIndexCache.java:407)
	at org.apache.kafka.storage.internals.log.RemoteIndexCache.loadIndexFile(RemoteIndexCache.java:354)
	at org.apache.kafka.storage.internals.log.RemoteIndexCache.createCacheEntry(RemoteIndexCache.java:403)
	at org.apache.kafka.storage.internals.log.RemoteIndexCache.lambda$getIndexEntry$10(RemoteIndexCache.java:377)
	at com.github.benmanes.caffeine.cache.BoundedLocalCache.lambda$doComputeIfAbsent$14(BoundedLocalCache.java:2406)
	at java.base/java.util.concurrent.ConcurrentHashMap.compute(ConcurrentHashMap.java:1947)
	at com.github.benmanes.caffeine.cache.BoundedLocalCache.doComputeIfAbsent(BoundedLocalCache.java:2404)
	at com.github.benmanes.caffeine.cache.BoundedLocalCache.computeIfAbsent(BoundedLocalCache.java:2387)
	at com.github.benmanes.caffeine.cache.LocalCache.computeIfAbsent(LocalCache.java:108)
	at com.github.benmanes.caffeine.cache.LocalManualCache.get(LocalManualCache.java:62)
	at org.apache.kafka.storage.internals.log.RemoteIndexCache.getIndexEntry(RemoteIndexCache.java:376)
	at kafka.log.remote.RemoteLogManager.lambda$collectAbortedTransactions$23(RemoteLogManager.java:1718)
	at java.base/java.util.Optional.map(Optional.java:265)
	at kafka.log.remote.RemoteLogManager.collectAbortedTransactions(RemoteLogManager.java:1718)
	at kafka.log.remote.RemoteLogManager.addAbortedTransactions(RemoteLogManager.java:1702)
	at kafka.log.remote.RemoteLogManager.read(RemoteLogManager.java:1666)
	at kafka.log.remote.RemoteLogReader.lambda$call$0(RemoteLogReader.java:66)
	at com.yammer.metrics.core.Timer.time(Timer.java:91)
	at kafka.log.remote.RemoteLogReader.call(RemoteLogReader.java:66)
	at kafka.log.remote.RemoteLogReader.call(RemoteLogReader.java:36)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: org.apache.kafka.server.log.remote.storage.RemoteStorageException: java.lang.RuntimeException: java.lang.InterruptedException
	at io.aiven.kafka.tieredstorage.RemoteStorageManager.fetchIndex(RemoteStorageManager.java:620)
	at org.apache.kafka.server.log.remote.storage.ClassLoaderAwareRemoteStorageManager.lambda$fetchIndex$5(ClassLoaderAwareRemoteStorageManager.java:88)
	at org.apache.kafka.server.log.remote.storage.ClassLoaderAwareRemoteStorageManager.withClassLoader(ClassLoaderAwareRemoteStorageManager.java:65)
	at org.apache.kafka.server.log.remote.storage.ClassLoaderAwareRemoteStorageManager.fetchIndex(ClassLoaderAwareRemoteStorageManager.java:88)
	at org.apache.kafka.storage.internals.log.RemoteIndexCache.lambda$createCacheEntry$13(RemoteIndexCache.java:405)
	... 23 more
Caused by: java.lang.RuntimeException: java.lang.InterruptedException
	at io.aiven.kafka.tieredstorage.fetch.index.MemorySegmentIndexesCache.get(MemorySegmentIndexesCache.java:132)
	at io.aiven.kafka.tieredstorage.RemoteStorageManager.fetchIndex(RemoteStorageManager.java:610)
	... 27 more
Caused by: java.lang.InterruptedException
	at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:385)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2028)
	at io.aiven.kafka.tieredstorage.fetch.index.MemorySegmentIndexesCache.get(MemorySegmentIndexesCache.java:110)
	... 28 more
DEBUG Deleted cached value for key SegmentIndexKey{indexesKey=wallet-events-KX2_hfceTraIuwXMZ0PQmA/6/00000000000103367174-mALEtIdVQXiVrv3l_fyGAg.indexes, indexType=TIMESTAMP} from cache. The reason of the deletion is SIZE (io.aiven.kafka.tieredstorage.fetch.index.MemorySegmentIndexesCache)
DEBUG Deleted cached value for key SegmentIndexKey{indexesKey=wallet-events-KX2_hfceTraIuwXMZ0PQmA/6/00000000000010536598-Bbi0rOhlTe2P1LQmzKOPDQ.indexes, indexType=OFFSET} from cache. The reason of the deletion is SIZE (io.aiven.kafka.tieredstorage.fetch.index.MemorySegmentIndexesCache)

The same message appears over and over, any thoughts?
Below is the configuration:
kafka 3.9.0 with S3 remote storage release 2024-10-23-1729694047

remote.log.storage.system.enable=true
remote.log.metadata.manager.class.name=org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager
remote.log.metadata.manager.listener.name=PLAINTEXT
remote.log.storage.manager.class.path=/opt/kafka/core/:/opt/kafka/s3/
remote.log.storage.manager.class.name=io.aiven.kafka.tieredstorage.RemoteStorageManager
rsm.config.chunk.size=4194304
rsm.config.storage.backend.class=io.aiven.kafka.tieredstorage.storage.s3.S3Storage
rsm.config.storage.s3.bucket.name=my-test-kafka-remote-tier
rsm.config.storage.s3.region=us-west-1
rsm.config.storage.s3.credentials.default=true
rsm.config.fetch.chunk.cache.class=io.aiven.kafka.tieredstorage.fetch.cache.DiskChunkCache
rsm.config.fetch.chunk.cache.path=/opt/kafka/rsm-cache
rsm.config.fetch.chunk.cache.size=98784247808
rsm.confi.fetch.chunk.cache.prefetch.max.size=16777216
rsm.config.fetch.chunk.cache.retention.ms=1200000
rsm.config.upload.rate.limit.bytes.per.second=104857600
remote.storage.enable=true

Thank you

[catalinmer]

And after a while it starts logging continuously (almost every milisecond) the following message:

ERROR Error occurred while reading the remote data for wallet-events-9 (kafka.log.remote.RemoteLogReader)
org.apache.kafka.common.KafkaException: Failed read position from the transaction index /var/lib/kafka-logs/remote-log-index-cache/73247352_p_apCq9bR5KPE7Jrw58UwQ.txnindex
at org.apache.kafka.storage.internals.log.TransactionIndex$1.hasNext(TransactionIndex.java:235)
at org.apache.kafka.storage.internals.log.TransactionIndex.collectAbortedTxns(TransactionIndex.java:171)
at kafka.log.remote.RemoteLogManager.collectAbortedTransactions(RemoteLogManager.java:1720)
at kafka.log.remote.RemoteLogManager.addAbortedTransactions(RemoteLogManager.java:1702)
at kafka.log.remote.RemoteLogManager.read(RemoteLogManager.java:1666)
at kafka.log.remote.RemoteLogReader.lambda$call$0(RemoteLogReader.java:66)
at com.yammer.metrics.core.Timer.time(Timer.java:91)
at kafka.log.remote.RemoteLogReader.call(RemoteLogReader.java:66)
at kafka.log.remote.RemoteLogReader.call(RemoteLogReader.java:36)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.nio.channels.ClosedChannelException
at java.base/sun.nio.ch.FileChannelImpl.ensureOpen(FileChannelImpl.java:150)
at java.base/sun.nio.ch.FileChannelImpl.position(FileChannelImpl.java:325)
at org.apache.kafka.storage.internals.log.TransactionIndex$1.hasNext(TransactionIndex.java:233)
... 12 more

[catalinmer]

I also noticed files in /var/lib/kafka-logs/remote-log-index-cache get replaced almost every second, and there are hundreds of them

-rw-r--r--.    1 kafka kafka 1854288 Jun 18 21:17 98140416_Wvq9wq7lQLGXi5A1675CyA.index
-rw-r--r--.    1 kafka kafka 2781444 Jun 18 21:17 98140416_Wvq9wq7lQLGXi5A1675CyA.timeindex
-rw-r--r--.    1 kafka kafka       0 Jun 18 21:17 98140416_Wvq9wq7lQLGXi5A1675CyA.txnindex
-rw-r--r--.    1 kafka kafka 1864680 Jun 18 21:17 98399052_9M-J3vkIR7K3okFreelwMA.index
-rw-r--r--.    1 kafka kafka 2797020 Jun 18 21:17 98399052_9M-J3vkIR7K3okFreelwMA.timeindex
-rw-r--r--.    1 kafka kafka       0 Jun 18 21:17 98399052_9M-J3vkIR7K3okFreelwMA.txnindex
-rw-r--r--.    1 kafka kafka 1853824 Jun 18 21:17 99067182_cj4w2Kh9T9Wnnav0I98jGg.index
-rw-r--r--.    1 kafka kafka 2780736 Jun 18 21:17 99067182_cj4w2Kh9T9Wnnav0I98jGg.timeindex
-rw-r--r--.    1 kafka kafka       0 Jun 18 21:17 99067182_cj4w2Kh9T9Wnnav0I98jGg.txnindex
-rw-r--r--.    1 kafka kafka 1860840 Jun 18 21:17 99339067_dZTfSgkNR-2rUnYxBiu_9Q.index
-rw-r--r--.    1 kafka kafka 2791272 Jun 18 21:17 99339067_dZTfSgkNR-2rUnYxBiu_9Q.timeindex

Also a lot of incoming traffic into all the servers of the cluster. Might be fetching from S3 something?
Normal incoming traffic is around 500Kbs and now it stays in 70Mbs.

[jeqo]

@catalinmer thanks for reporting this. The remote fetch thread may get interrupted because of the remote fetch timeout1.
Could you give a try and increase the timeout (e.g. to 1 or 2sec) and check the results?

[catalinmer]

Thank you for the reply. Unfortunately it doesn't help. I changed it but the error messages are still there. What is really strange is that we have no consumer currently consuming historical data from that topic, but each kafka broker still fetches about 60Mbs of content from S3. There is no consumer lag and we move data to S3 after 30 days or 10G.

• local.retention.ms=2592000000
• local.retention.bytes=10737418240
  So it is a mystery what kafka tries to fetch. The content of /var/lib/kafka-logs/remote-log-index-cache/ changes almost entirely every second. I'm trying to decode some index files maybe I can figure out what it tries to fetch.
  I also enabled debug for log4j.logger.io.aiven.kafka.tieredstorage=DEBUG but other than a million lines of

Deleted cached value for key SegmentIndexKey{indexesKey=....} from cache. The reason of the deletion is SIZE (io.aiven.kafka.tieredstorage.fetch.index.MemorySegmentIndexesCache)

and a few of

DEBUG Deleted cached value for key wallet-events-KX2_hfceTraIuwXMZ0PQmA/3/00000000000000000001-lw6hopdFQEWN2kaecfYo5Q.rsm-manifest from cache. The reason of the deletion is EXPIRED (io.aiven.kafka.tieredstorage.fetch.manifest.MemorySegmentManifestCache)

there is nothing.

[catalinmer]

Ok, I found the problem. It was due to the default low value of remote.log.index.file.cache.total.size.bytes. It might be a bug as well related to this behavior.
What I noticed was that it tried to fetch from remote the index files on one hand, while it was deleting them (due to cache size limit - 1GB) on the other and it was doing that forever without actually fetching a single entry from the topic.

[jeqo]

The cache on the broker side has some limitations AFAIR. You could also try to disable it by setting a small size, and rely on the plugin's cache for indexes.

[DDDFiish]

I encountered the same issue. I would like to ask how you resolved it. Did you increase the value of remote.log.index.file.cache.total.size.bytes?