refactor(inkless): add artificial latency to metadata update

When refreshing metadata, new requests to next broker may be processed
_before_ the last request on the previous broker, causing
OutOfOrderSequence.
These errors are retried and data finally lands in order.
However to avoid this error, this proposal adds artificial latency to
the metadata response when it includes inkless topics.
This way it gives enough time for the previous broker to complete
processing requests, before the new one takes on.

Author: jeqo

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

@@ -125,7 +125,7 @@ class KafkaApis(val requestChannel: RequestChannel,
   val describeTopicPartitionsRequestHandler = new DescribeTopicPartitionsRequestHandler(
     metadataCache, authHelper, config)
 
-  val inklessTopicMetadataTransformer = inklessSharedState.map(s => new InklessTopicMetadataTransformer(s.metadata()))
+  val inklessTopicMetadataTransformer = inklessSharedState.map(s => new InklessTopicMetadataTransformer(time, s.metadata()))
 
   def close(): Unit = {
     aclApis.close()

storage/inkless/src/main/java/io/aiven/inkless/metadata/InklessTopicMetadataTransformer.java

@@ -21,6 +21,7 @@
 import org.apache.kafka.common.message.DescribeTopicPartitionsResponseData;
 import org.apache.kafka.common.message.MetadataResponseData;
 import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.metadata.LeaderAndIsr;
 
 import java.util.Collections;
@@ -33,11 +34,13 @@
 import io.aiven.inkless.control_plane.MetadataView;
 
 public class InklessTopicMetadataTransformer {
+    private final Time time;
     private final MetadataView metadataView;
 
     private final AtomicInteger roundRobinCounter = new AtomicInteger();
 
-    public InklessTopicMetadataTransformer(final MetadataView metadataView) {
+    public InklessTopicMetadataTransformer(final Time time, final MetadataView metadataView) {
+        this.time = Objects.requireNonNull(time, "time cannot be null");
         this.metadataView = Objects.requireNonNull(metadataView, "metadataView cannot be null");
     }
 
@@ -52,10 +55,12 @@ public void transformClusterMetadata(
         Objects.requireNonNull(topicMetadata, "topicMetadata cannot be null");
 
         final int leaderForInklessPartitions = selectLeaderForInklessPartitions(listenerName, clientId);
+        boolean hasInklessTopics = false;
         for (final var topic : topicMetadata) {
             if (!metadataView.isInklessTopic(topic.name())) {
                 continue;
             }
+            hasInklessTopics = true;
             for (final var partition : topic.partitions()) {
                 partition.setLeaderId(leaderForInklessPartitions);
                 final List<Integer> list = List.of(leaderForInklessPartitions);
@@ -65,6 +70,11 @@ public void transformClusterMetadata(
                 partition.setLeaderEpoch(LeaderAndIsr.INITIAL_LEADER_EPOCH);
             }
         }
+        if (hasInklessTopics) {
+            // Introduce artificial latency to avoid a race condition between the metadata update and the producer
+            // causing OutOfOrderSequenceException.
+            time.sleep(500);
+        }
     }
 
     /**

storage/inkless/src/test/java/io/aiven/inkless/metadata/InklessTopicMetadataTransformerTest.java

@@ -28,6 +28,8 @@
 import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
 
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
@@ -62,16 +64,20 @@ class InklessTopicMetadataTransformerTest {
     static final Uuid TOPIC_CLASSIC_ID = new Uuid(456, 456);
     static final ListenerName LISTENER_NAME = ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT);
 
+    Time time = new MockTime();
     @Mock
     MetadataView metadataView;
 
     @Test
     void nulls() {
-        assertThatThrownBy(() -> new InklessTopicMetadataTransformer(null))
+        assertThatThrownBy(() -> new InklessTopicMetadataTransformer(null, metadataView))
+            .isInstanceOf(NullPointerException.class)
+            .hasMessage("time cannot be null");
+        assertThatThrownBy(() -> new InklessTopicMetadataTransformer(time, null))
             .isInstanceOf(NullPointerException.class)
             .hasMessage("metadataView cannot be null");
 
-        final var transformer = new InklessTopicMetadataTransformer(metadataView);
+        final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
         assertThatThrownBy(() -> transformer.transformClusterMetadata(LISTENER_NAME, "x", null))
             .isInstanceOf(NullPointerException.class)
             .hasMessage("topicMetadata cannot be null");
@@ -94,7 +100,7 @@ void setup() {
         @NullSource
         @ValueSource(strings = {"inkless_az=az1", "x=y", ""})
         void clusterMetadata(final String clientId) {
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
 
             final List<MetadataResponseTopic> topicMetadata = List.of();
             transformer.transformClusterMetadata(LISTENER_NAME, clientId, topicMetadata);
@@ -105,7 +111,7 @@ void clusterMetadata(final String clientId) {
         @NullSource
         @ValueSource(strings = {"inkless_az=az1", "x=y", ""})
         void describeTopicResponse(final String clientId) {
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
 
             final DescribeTopicPartitionsResponseData describeResponse = new DescribeTopicPartitionsResponseData();
             transformer.transformDescribeTopicResponse(LISTENER_NAME, clientId, describeResponse);
@@ -186,7 +192,7 @@ void clusterMetadata(final String clientAZ, final int expectedLeaderId1, final i
                 inklessTopicMetadata.get(),
                 classicTopicMetadata.get()
             );
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
 
             transformer.transformClusterMetadata(LISTENER_NAME, "inkless_az=" + clientAZ, topicMetadata);
 
@@ -279,7 +285,7 @@ void describeTopicResponse(final String clientAZ, final int expectedLeaderId1, f
                         inklessTopicMetadata.get(),
                         classicTopicMetadata.get()
                     ).iterator()));
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
 
             transformer.transformDescribeTopicResponse(LISTENER_NAME, "inkless_az=" + clientAZ, describeResponse);
 
@@ -333,7 +339,7 @@ void clusterMetadata() {
                     ));
 
             final List<MetadataResponseTopic> topicMetadata = List.of(inklessTopicMetadata.get());
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
 
             transformer.transformClusterMetadata(LISTENER_NAME, "inkless_az=az0", topicMetadata);
             final var expectedInklessTopicMetadata = inklessTopicMetadata.get();
@@ -368,7 +374,7 @@ void describeTopicResponse() {
                             ))
                     ).iterator()));
 
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
 
             final DescribeTopicPartitionsResponseData describeResponse = describeResponseSupplier.get();
             transformer.transformDescribeTopicResponse(LISTENER_NAME, "inkless_az=az0", describeResponse);
@@ -413,7 +419,7 @@ void clusterMetadata() {
                     ));
 
             final List<MetadataResponseTopic> topicMetadata = List.of(inklessTopicMetadata.get());
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
 
             transformer.transformClusterMetadata(LISTENER_NAME, null, topicMetadata);
             final var expectedInklessTopicMetadata = inklessTopicMetadata.get();
@@ -448,7 +454,7 @@ void describeTopicResponse() {
                             ))
                     ).iterator()));
 
-            final var transformer = new InklessTopicMetadataTransformer(metadataView);
+            final var transformer = new InklessTopicMetadataTransformer(time, metadataView);
             final DescribeTopicPartitionsResponseData describeResponse = describeResponseSupplier.get();
 
             transformer.transformDescribeTopicResponse(LISTENER_NAME, null, describeResponse);