cleanup

Author: giuseppelillo

core/src/main/scala/kafka/server/DelayedFetch.scala

@@ -159,13 +159,10 @@ class DelayedFetch(
     }
 
     // Case G
-    if (accumulatedSize >= minBytes.getOrElse(params.minBytes)) {
-      DelayedFetchMetrics.enoughBytesMeter.mark()
+    if (accumulatedSize >= minBytes.getOrElse(params.minBytes))
       forceComplete()
-    } else {
-      DelayedFetchMetrics.notEnoughBytesMeter.mark()
+    else
       false
-    }
   }
 
   /**
@@ -186,12 +183,10 @@ class DelayedFetch(
     if (requests.isEmpty) return Some(0)
 
     val response = try {
-      DelayedFetchMetrics.findBatchesRequestsMeter.mark()
       replicaManager.findDisklessBatches(requests)
     } catch {
       case e: Throwable =>
         error("Error while trying to find diskless batches on delayed fetch.", e)
-        DelayedFetchMetrics.findBatchesErrorsMeter.mark()
         return None  // Case C
     }
 
@@ -206,7 +201,6 @@ class DelayedFetch(
             val fetchPartitionStatus = fetchPartitionStatusMap.get(topicIdPartition)
             if (fetchPartitionStatus.isEmpty) {
               warn(s"Fetch partition status for $topicIdPartition not found in delayed fetch $this.")
-              DelayedFetchMetrics.findBatchesErrorsMeter.mark()
               return None  // Case C
             }
 
@@ -217,20 +211,13 @@ class DelayedFetch(
             if (fetchOffset.messageOffset > endOffset) {
               // Truncation happened
               debug(s"Satisfying fetch $this since it is fetching later segments of partition $topicIdPartition.")
-              DelayedFetchMetrics.logTruncationsMeter.mark()
               return None  // Case A
             } else if (fetchOffset.messageOffset < endOffset) {
               val bytesAvailable = r.estimatedByteSize(fetchOffset.messageOffset)
               accumulatedSize += bytesAvailable // Case B: accumulate the size of the batches
-            } else {
-              // Case D: same as fetchOffset == endOffset, no new data available
-              DelayedFetchMetrics.noNewDataMeter.mark()
-            }
+            } // Case D: same as fetchOffset == endOffset, no new data available
           }
-        case _ => {
-          DelayedFetchMetrics.findBatchesErrorsMeter.mark()
-          return None
-        }  // Case C
+        case _ => return None  // Case C
       }
     }
 
@@ -310,12 +297,5 @@ object DelayedFetchMetrics {
   private val FetcherTypeKey = "fetcherType"
   val followerExpiredRequestMeter: Meter = metricsGroup.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS, Map(FetcherTypeKey -> "follower").asJava)
   val consumerExpiredRequestMeter: Meter = metricsGroup.newMeter("ExpiresPerSec", "requests", TimeUnit.SECONDS, Map(FetcherTypeKey -> "consumer").asJava)
-  // diskless metrics
-  val findBatchesRequestsMeter: Meter = metricsGroup.newMeter("FindBatchesRequestsPerSec", "requests", TimeUnit.SECONDS)
-  val findBatchesErrorsMeter: Meter = metricsGroup.newMeter("FindBatchesErrorsPerSec", "requests", TimeUnit.SECONDS)
-  val enoughBytesMeter: Meter = metricsGroup.newMeter("EnoughBytesPerSec", "requests", TimeUnit.SECONDS)
-  val logTruncationsMeter: Meter = metricsGroup.newMeter("LogTruncationsPerSec", "requests", TimeUnit.SECONDS)
-  val noNewDataMeter: Meter = metricsGroup.newMeter("NoNewDataPerSec", "requests", TimeUnit.SECONDS)
-  val notEnoughBytesMeter: Meter = metricsGroup.newMeter("NotEnoughBytesPerSec", "requests", TimeUnit.SECONDS)
 }
 

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -1727,7 +1727,7 @@ class ReplicaManager(val config: KafkaConfig,
 
   def findDisklessBatches(requests: Seq[FindBatchRequest]): Option[util.List[FindBatchResponse]] = {
     inklessSharedState.flatMap { sharedState =>
-      if (!sharedState.isBatchCoordinatesCacheEnabled) {
+      if (!sharedState.isBatchCoordinateCacheEnabled) {
         findDisklessBatchesThroughControlPlane(requests)
       } else {
         Some(requests.map { request =>

core/src/test/java/kafka/server/InklessClusterTest.java

@@ -37,8 +37,6 @@
 import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.serialization.IntegerDeserializer;
-import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.test.KafkaClusterTestKit;
 import org.apache.kafka.common.test.TestKitNodes;
 import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
@@ -57,7 +55,6 @@
 import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -192,81 +189,6 @@ public void createDisklessTopic(final boolean idempotenceEnable, final Timestamp
         consumeWithSubscription(timestampType, clientConfigs, topicName, now, numRecords);
     }
 
-    @Test
-    public void test() throws Exception {
-        Map<String, Object> clientConfigs = new HashMap<>();
-        clientConfigs.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
-        clientConfigs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, String.valueOf(true));
-        clientConfigs.put(ProducerConfig.LINGER_MS_CONFIG, "1000");
-        clientConfigs.put(ProducerConfig.BATCH_SIZE_CONFIG, "100000");
-        clientConfigs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-        clientConfigs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
-        clientConfigs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
-        clientConfigs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class.getName());
-        // by default is latest and nothing would get consumed.
-        clientConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, AutoOffsetResetStrategy.EARLIEST.name());
-        String topicName = "inkless-cache-topic";
-        int numRecords = 50;
-
-        try (Admin admin = AdminClient.create(clientConfigs)) {
-            final NewTopic topic = new NewTopic(topicName, 2, (short) 1)
-                .configs(Map.of(
-                    TopicConfig.DISKLESS_ENABLE_CONFIG, "true",
-                    TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, TimestampType.LOG_APPEND_TIME.name
-                ));
-            CreateTopicsResult topics = admin.createTopics(Collections.singletonList(topic));
-            topics.all().get(10, TimeUnit.SECONDS);
-        }
-
-        AtomicInteger recordsProduced = new AtomicInteger();
-        final long now = System.currentTimeMillis();
-        Callback produceCb = (metadata, exception) -> {
-            if (exception != null) {
-                log.error("Failed to send record", exception);
-            } else {
-                log.info("Committed value for topic {} at offset {} at {}", metadata.topic(), metadata.offset(), now);
-                recordsProduced.incrementAndGet();
-            }
-        };
-        var producedRecords1 = new ArrayList<Integer>(numRecords);
-        var producedRecords2 = new ArrayList<Integer>(numRecords);
-        try (Producer<byte[], Integer> producer = new KafkaProducer<>(clientConfigs)) {
-            for (int i = 0; i < numRecords; i++) {
-                producedRecords1.add(i);
-                final ProducerRecord<byte[], Integer> record1 = new ProducerRecord<>(topicName, 0, now, null, i);
-                producedRecords2.add(i + 10000);
-                final ProducerRecord<byte[], Integer> record2 = new ProducerRecord<>(topicName, 1, now, null, i + 10000);
-                producer.send(record1, produceCb);
-                Thread.sleep(100);
-                producer.send(record2, produceCb);
-                Thread.sleep(100);
-            }
-            producer.flush();
-        }
-
-        var consumedRecords1 = new ArrayList<Integer>();
-        var consumedRecords2 = new ArrayList<Integer>();
-        assertEquals(numRecords * 2, recordsProduced.get());
-        try (Consumer<byte[], Integer> consumer = new KafkaConsumer<>(clientConfigs)) {
-            consumer.assign(Collections.singletonList(new TopicPartition(topicName, 0)));
-            ConsumerRecords<byte[], Integer> poll = consumer.poll(Duration.ofSeconds(30));
-            for (ConsumerRecord<byte[], Integer> record : poll) {
-                assertTrue(record.timestamp() > now);
-                consumedRecords1.add(record.value());
-            }
-        }
-        try (Consumer<byte[], Integer> consumer = new KafkaConsumer<>(clientConfigs)) {
-            consumer.assign(Collections.singletonList(new TopicPartition(topicName, 1)));
-            ConsumerRecords<byte[], Integer> poll = consumer.poll(Duration.ofSeconds(30));
-            for (ConsumerRecord<byte[], Integer> record : poll) {
-                assertTrue(record.timestamp() > now);
-                consumedRecords2.add(record.value());
-            }
-        }
-        assertEquals(producedRecords1, consumedRecords1);
-        assertEquals(producedRecords2, consumedRecords2);
-    }
-
     @Test
     public void produceToDisklessAndClassic() throws Exception {
         Map<String, Object> clientConfigs = new HashMap<>();

docker/examples/docker-compose-files/inkless/docker-compose.yml

@@ -27,10 +27,7 @@ services:
       # Inkless
       KAFKA_DISKLESS_STORAGE_SYSTEM_ENABLE: "true"
       KAFKA_LOG_DISKLESS_ENABLE: "true"
-      KAFKA_INKLESS_CONSUME_BATCH_COORDINATES_CACHE_ENABLED: "false"
-      KAFKA_INKLESS_CONSUME_BATCH_COORDINATES_CACHE_TTL_MS: "5000"
       KAFKA_INKLESS_CONSUME_CACHE_MAX_COUNT: "100"
-      KAFKA_INKLESS_CONSUME_CACHE_EXPIRATION_MAX_IDLE_SEC: "30"
 
       ## Control Plane
       ### Postgresql
@@ -47,7 +44,7 @@ services:
 
       # JMX
       KAFKA_JMX_PORT: 9999
-      KAFKA_HEAP_OPTS: "-Xmx8G -Xms8G"
+      KAFKA_HEAP_OPTS: "-Xmx2G -Xms2G"
       EXTRA_ARGS: -javaagent:/opt/prometheus-jmx-exporter/jmx_prometheus_javaagent.jar=7070:/opt/prometheus-jmx-exporter/kafka.yml
     volumes:
       - ./../../../extra/prometheus-jmx-exporter:/opt/prometheus-jmx-exporter:Z

docs/inkless/configs.rst

@@ -63,15 +63,15 @@ Under ``inkless.``
   * Valid Values: [0,...]
   * Importance: medium
 
-``consume.batch.coordinates.cache.enabled``
-  If true, batch coordinates cache is enabled.
+``consume.batch.coordinate.cache.enabled``
+  If true, the Batch Coordinate cache is enabled.
 
   * Type: boolean
   * Default: true
   * Importance: low
 
-``consume.batch.coordinates.cache.ttl.ms``
-  Time to live in milliseconds for an entry in the Batch Coordinates cache. The time to live must be <= than half of the value of of file.cleaner.interval.ms.
+``consume.batch.coordinate.cache.ttl.ms``
+  Time to live in milliseconds for an entry in the Batch Coordinate cache. The time to live must be <= than half of the value of of file.cleaner.interval.ms.
 
   * Type: int
   * Default: 5000 (5 seconds)

storage/inkless/src/main/java/io/aiven/inkless/cache/CacheBatchCoordinate.java

@@ -24,7 +24,7 @@
 import io.aiven.inkless.control_plane.BatchMetadata;
 
 /**
- * Represents the coordinates of batch that is handled by the Batch Coordinate cache
+ * Represents the coordinates of a batch that is handled by the Batch Coordinate cache
  */
 public record CacheBatchCoordinate(
     TopicIdPartition topicIdPartition,

storage/inkless/src/main/java/io/aiven/inkless/cache/CaffeineBatchCoordinateCache.java

@@ -13,6 +13,12 @@
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicInteger;
 
+/**
+ * A thread-safe cache that manages {@link LogFragment} instances for
+ * multiple Kafka diskless topic partitions.
+ * It uses a time-to-live (TTL) parameter as a cleanup policy.
+ * This cache automatically handles stale and expired entries.
+ */
 public class CaffeineBatchCoordinateCache implements BatchCoordinateCache {
 
     static class UncheckedStaleCacheEntryException extends RuntimeException {
@@ -57,6 +63,25 @@ public CaffeineBatchCoordinateCache(Duration ttl) {
     }
 
 
+    /**
+     * Retrieves a view of the log's tail for a specific partition, starting from
+     * the requested offset.
+     *
+     * <p>This operation will result in a <b>cache miss</b> (returning {@code null})
+     * in several scenarios:
+     * <ul>
+     * <li>If the entire partition is not currently tracked by the cache.</li>
+     * <li>If the requested {@code offset} is older than the data held in the
+     * fragment (i.e., it has already been evicted).</li>
+     * <li>If the data at that offset is found but has expired at the moment of
+     * the request.</li>
+     * </ul></p>
+     *
+     * @param topicIdPartition The specific topic partition to read from.
+     * @param offset           The desired starting offset for the read.
+     * @return A new {@code LogFragment} view containing the requested data, or
+     * {@code null} if the data is not available (a cache miss).
+     */
     @Override
     public LogFragment get(TopicIdPartition topicIdPartition, long offset) {
         final LogFragment logFragment = cache.getIfPresent(topicIdPartition);
@@ -65,7 +90,14 @@ public LogFragment get(TopicIdPartition topicIdPartition, long offset) {
             return null;
         }
         metrics.recordCacheHit();
-        return logFragment.subFragment(offset);
+        final LogFragment subFragment = logFragment.subFragment(offset);
+        if (subFragment == null || subFragment.isEmpty()) {
+            metrics.recordCacheMiss();
+        } else {
+            metrics.recordCacheHit();
+        }
+
+        return subFragment;
     }
 
     private void putInternal(CacheBatchCoordinate value) throws StaleCacheEntryException {
@@ -103,6 +135,24 @@ private void evictExpiredEntries(LogFragment logFragment) {
         }
     }
 
+    /**
+     * Appends a new batch coordinate to the cache, adding it to the correct log
+     * fragment for its partition.
+     *
+     * <p>This method handles stale cache entries: when a batch is added,
+     * it is passed to the appropriate {@link LogFragment}.
+     * If that fragment detects that the batch would create an integrity problem
+     * (such as a data gap, an overlap, or if the batch indicates the log start offset was
+     * increased), the entire cache entry for the requested TopicPartition will be invalidated
+     * and the batch insertion will be retried, creating a new log fragment.</p>
+     *
+     * <p>After a successful batch addition, this method also opportunistically
+     * triggers an eviction check on the fragment to purge any expired batches.</p>
+     *
+     * @param cacheBatchCoordinate The batch coordinate to add.
+     * @throws IllegalStateException If the new batch is rejected for being invalid (lower log start offset
+     * than the one present in the cache, or lower base offset than the high watermark present in the cache)
+     */
     @Override
     public void put(CacheBatchCoordinate cacheBatchCoordinate) throws IllegalStateException {
         TopicIdPartition topicIdPartition = cacheBatchCoordinate.topicIdPartition();
@@ -126,7 +176,9 @@ public void put(CacheBatchCoordinate cacheBatchCoordinate) throws IllegalStateEx
     }
 
     /**
-     * Removes the entry from the cache for a topic partition
+     * Removes the entry from the cache for a topic partition.
+     * @param topicIdPartition The specific topic partition to remove from the cache.
+     * @return the number of batches removed from the cache.
      */
     @Override
     public int invalidatePartition(TopicIdPartition topicIdPartition) {