fix(inkless:consolidation): prevent ConsolidationFetcherThread crash on topic deletion (#627)

When a diskless topic is deleted, the ConsolidationFetcherThread crashes
permanently in buildFetch():
- localLogOrException() throws UnknownTopicOrPartitionException
- only KafkaStorageException is caught ? exception escapes doWork()
- ShutdownableThread exits with no recovery

This kills consolidation for ALL topics on the broker until restart.

The race: BrokerMetadataPublisher sets metadataCache.setImage(newImage)
(removing the topic from the cache) before applyDelta calls
stopPartitions (which removes the partition from the fetcher). In that
window, the fetcher's buildFetch reads a partition still in
partitionStates but gone from allPartitions/metadataCache.

The regular ReplicaFetcher never hits this because stopPartitions removes
from the fetcher BEFORE allPartitions.remove Ñ so the fetcher never sees
a partition that's been deleted from allPartitions. The consolidation
fetcher hits the window because metadata-view staleness creates a gap.

Fix: catch UnknownTopicOrPartitionException alongside KafkaStorageException
in DisklessLeaderEndPoint.buildFetch. The partition is added to
partitionsWithError, delayed, and evicted on the next cycle once
removePartitions completes.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: jeqo

.github/workflows/inkless.yml

@@ -169,11 +169,13 @@ jobs:
             --tests "org.apache.kafka.image.TopicsImageTest" && \
           ./gradlew ${GRADLE_ARGS} :core:test \
             --tests "*Inkless*" --tests "*Diskless*" \
+            --tests "io.aiven.inkless.*" \
             --tests "kafka.log.LogConfigTest" \
             --tests "kafka.server.KafkaConfigTest" \
             --tests "kafka.server.ReplicaManagerTest" \
             --tests "kafka.server.KafkaApisTest" \
             --tests "kafka.cluster.PartitionTest" \
+            --tests "kafka.server.AbstractFetcherThreadTest" \
             --tests "kafka.server.DynamicBrokerConfigTest" \
             --tests "kafka.server.DynamicConfigChangeTest" \
             --tests "kafka.server.ReplicaFetcherThreadTest" \

core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala

@@ -22,7 +22,7 @@ import io.aiven.inkless.consume.{FetchHandler, FetchOffsetHandler}
 import kafka.server.{KafkaConfig, ReplicaManager, ReplicaQuota}
 import kafka.utils.Logging
 import org.apache.kafka.common.Uuid
-import org.apache.kafka.common.errors.KafkaStorageException
+import org.apache.kafka.common.errors.{KafkaStorageException, UnknownTopicOrPartitionException}
 import org.apache.kafka.common.message.{FetchResponseData, OffsetForLeaderEpochRequestData}
 import org.apache.kafka.common.message.ListOffsetsRequestData.ListOffsetsPartition
 import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.EpochEndOffset
@@ -251,7 +251,10 @@ class DisklessLeaderEndPoint(
               )
             )
           } catch {
-            case _: KafkaStorageException =>
+            // UnknownTopicOrPartitionException from localLogOrException when partition is
+            // deleted from allPartitions before the fetcher's partitionStates is cleaned up.
+            case e @ (_: KafkaStorageException | _: UnknownTopicOrPartitionException) =>
+              logger.info("Partition {} unavailable during buildFetch: {}", topicPartition, e.getMessage)
               partitionsWithError += topicPartition
           }
         }

core/src/test/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPointTest.scala

@@ -462,6 +462,29 @@ class DisklessLeaderEndPointTest {
     assertEquals(Set(topicPartition), result.partitionsWithError.asScala.toSet)
   }
 
+  @Test
+  def testBuildFetchMarksPartitionWithUnknownTopicOrPartitionException(): Unit = {
+    val fetchHandler = mock(classOf[FetchHandler])
+    val fetchOffsetHandler = mock(classOf[FetchOffsetHandler])
+    val replicaManager = mock(classOf[ReplicaManager])
+    when(replicaManager.localLogOrException(topicPartition)).thenThrow(new UnknownTopicOrPartitionException("deleted"))
+
+    val endPoint = newEndPoint(fetchHandler, fetchOffsetHandler, replicaManager)
+    val fetchState = new PartitionFetchState(
+      Optional.of(topicId),
+      0L,
+      Optional.empty(),
+      1,
+      Optional.empty(),
+      ReplicaState.FETCHING,
+      Optional.empty()
+    )
+    val result = endPoint.buildFetch(util.Map.of(topicPartition, fetchState))
+
+    assertTrue(result.result.isEmpty)
+    assertEquals(Set(topicPartition), result.partitionsWithError.asScala.toSet)
+  }
+
   @Test
   def testBuildFetchSkipsPartitionWhenFollowerShouldThrottle(): Unit = {
     val fetchHandler = mock(classOf[FetchHandler])