KAFKA-12376: Apply atomic append to the log

core/src/main/scala/kafka/raft/RaftManager.scala

@@ -91,6 +91,11 @@ trait RaftManager[T] {
     listener: RaftClient.Listener[T]
   ): Unit
 
+  def scheduleAtomicAppend(
+    epoch: Int,
+    records: Seq[T]
+  ): Option[Long]
+
   def scheduleAppend(
     epoch: Int,
     records: Seq[T]
@@ -156,16 +161,32 @@ class KafkaRaftManager[T](
     raftClient.register(listener)
   }
 
+  override def scheduleAtomicAppend(
+    epoch: Int,
+    records: Seq[T]
+  ): Option[Long] = {
+    append(epoch, records, true)
+  }
+
   override def scheduleAppend(
     epoch: Int,
     records: Seq[T]
   ): Option[Long] = {
-    val offset: java.lang.Long = raftClient.scheduleAppend(epoch, records.asJava)
-    if (offset == null) {
-      None
+    append(epoch, records, false)
+  }
+
+  private def append(
+    epoch: Int,
+    records: Seq[T],
+    isAtomic: Boolean
+  ): Option[Long] = {
+    val offset = if (isAtomic) {
+      raftClient.scheduleAtomicAppend(epoch, records.asJava)
     } else {
-      Some(Long.unbox(offset))
+      raftClient.scheduleAppend(epoch, records.asJava)
     }
+
+    Option(offset).map(Long.unbox)
   }
 
   override def handleRequest(

metadata/src/main/java/org/apache/kafka/controller/ClientQuotaControlManager.java

@@ -86,7 +86,7 @@ ControllerResult<Map<ClientQuotaEntity, ApiError>> alterClientQuotas(
             }
         });
 
-        return new ControllerResult<>(outputRecords, outputResults);
+        return new ControllerResult<>(outputRecords, outputResults, true);
     }
 
     /**

metadata/src/main/java/org/apache/kafka/controller/ConfigurationControlManager.java

@@ -83,7 +83,7 @@ ControllerResult<Map<ConfigResource, ApiError>> incrementalAlterConfigs(
                 outputRecords,
                 outputResults);
         }
-        return new ControllerResult<>(outputRecords, outputResults);
+        return new ControllerResult<>(outputRecords, outputResults, true);
     }
 
     private void incrementalAlterConfigResource(ConfigResource configResource,
@@ -171,7 +171,7 @@ ControllerResult<Map<ConfigResource, ApiError>> legacyAlterConfigs(
                 outputRecords,
                 outputResults);
         }
-        return new ControllerResult<>(outputRecords, outputResults);
+        return new ControllerResult<>(outputRecords, outputResults, true);
     }
 
     private void legacyAlterConfigResource(ConfigResource configResource,

metadata/src/main/java/org/apache/kafka/controller/ControllerResult.java

@@ -28,15 +28,21 @@
 class ControllerResult<T> {
     private final List<ApiMessageAndVersion> records;
     private final T response;
+    private final boolean isAtomic;
 
     public ControllerResult(T response) {
         this(new ArrayList<>(), response);
     }
 
     public ControllerResult(List<ApiMessageAndVersion> records, T response) {
+        this(records, response, false);
+    }
+
+    public ControllerResult(List<ApiMessageAndVersion> records, T response, boolean isAtomic) {
         Objects.requireNonNull(records);
         this.records = records;
         this.response = response;
+        this.isAtomic = isAtomic;
     }
 
     public List<ApiMessageAndVersion> records() {
@@ -47,29 +53,37 @@ public T response() {
         return response;
     }
 
+    public boolean isAtomic() {
+        return isAtomic;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (o == null || (!o.getClass().equals(getClass()))) {
             return false;
         }
         ControllerResult other = (ControllerResult) o;
         return records.equals(other.records) &&
-            Objects.equals(response, other.response);
+            Objects.equals(response, other.response) &&
+            Objects.equals(isAtomic, other.isAtomic);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(records, response);
+        return Objects.hash(records, response, isAtomic);
     }
 
     @Override
     public String toString() {
-        return "ControllerResult(records=" + String.join(",",
-            records.stream().map(r -> r.toString()).collect(Collectors.toList())) +
-            ", response=" + response + ")";
+        return String.format(
+            "ControllerResult(records=%s, response=%s, isAtomic=%s)",
+            String.join(",", records.stream().map(r -> r.toString()).collect(Collectors.toList())),
+            response,
+            isAtomic
+        );
     }
 
     public ControllerResult<T> withoutRecords() {
-        return new ControllerResult<>(new ArrayList<>(), response);
+        return new ControllerResult<>(new ArrayList<>(), response, false);
     }
 }

metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java

@@ -69,7 +69,7 @@ ControllerResult<Map<String, ApiError>> updateFeatures(
             results.put(entry.getKey(), updateFeature(entry.getKey(), entry.getValue(),
                 downgradeables.contains(entry.getKey()), brokerFeatures, records));
         }
-        return new ControllerResult<>(records, results);
+        return new ControllerResult<>(records, results, records.isEmpty() ? false : true);
     }
 
     private ApiError updateFeature(String featureName,

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -265,7 +265,7 @@ private Throwable handleEventException(String name,
     class ControlEvent implements EventQueue.Event {
         private final String name;
         private final Runnable handler;
-        private long eventCreatedTimeNs = time.nanoseconds();
+        private final long eventCreatedTimeNs = time.nanoseconds();
         private Optional<Long> startProcessingTimeNs = Optional.empty();
 
         ControlEvent(String name, Runnable handler) {
@@ -307,7 +307,7 @@ class ControllerReadEvent<T> implements EventQueue.Event {
         private final String name;
         private final CompletableFuture<T> future;
         private final Supplier<T> handler;
-        private long eventCreatedTimeNs = time.nanoseconds();
+        private final long eventCreatedTimeNs = time.nanoseconds();
         private Optional<Long> startProcessingTimeNs = Optional.empty();
 
         ControllerReadEvent(String name, Supplier<T> handler) {
@@ -389,7 +389,7 @@ class ControllerWriteEvent<T> implements EventQueue.Event, DeferredEvent {
         private final String name;
         private final CompletableFuture<T> future;
         private final ControllerWriteOperation<T> op;
-        private long eventCreatedTimeNs = time.nanoseconds();
+        private final long eventCreatedTimeNs = time.nanoseconds();
         private Optional<Long> startProcessingTimeNs = Optional.empty();
         private ControllerResultAndOffset<T> resultAndOffset;
 
@@ -441,7 +441,12 @@ public void run() throws Exception {
                 // written before we can return our result to the user.  Here, we hand off
                 // the batch of records to the metadata log manager.  They will be written
                 // out asynchronously.
-                long offset = logManager.scheduleWrite(controllerEpoch, result.records());
+                final long offset;
+                if (result.isAtomic()) {
+                    offset = logManager.scheduleAtomicWrite(controllerEpoch, result.records());
+                } else {
+                    offset = logManager.scheduleWrite(controllerEpoch, result.records());
+                }
                 op.processBatchEndOffset(offset);
                 writeOffset = offset;
                 resultAndOffset = new ControllerResultAndOffset<>(offset,

metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java

@@ -407,7 +407,7 @@ public void replay(PartitionChangeRecord record) {
             resultsPrefix = ", ";
         }
         log.info("createTopics result(s): {}", resultsBuilder.toString());
-        return new ControllerResult<>(records, data);
+        return new ControllerResult<>(records, data, true);
     }
 
     private ApiError createTopic(CreatableTopic topic,

metadata/src/main/java/org/apache/kafka/metalog/LocalLogManager.java

@@ -328,8 +328,21 @@ public void register(MetaLogListener listener) throws Exception {
 
     @Override
     public long scheduleWrite(long epoch, List<ApiMessageAndVersion> batch) {
-        return shared.tryAppend(nodeId, leader.epoch(), new LocalRecordBatch(
-            batch.stream().map(r -> r.message()).collect(Collectors.toList())));
+        return scheduleAtomicWrite(epoch, batch);
+    }
+
+    @Override
+    public long scheduleAtomicWrite(long epoch, List<ApiMessageAndVersion> batch) {
+        return shared.tryAppend(
+            nodeId,
+            leader.epoch(),
+            new LocalRecordBatch(
+                batch
+                    .stream()
+                    .map(r -> r.message())
+                    .collect(Collectors.toList())
+            )
+        );
     }
 
     @Override

metadata/src/main/java/org/apache/kafka/metalog/MetaLogManager.java

@@ -50,13 +50,30 @@ public interface MetaLogManager {
      * offset before renouncing its leadership.  The listener should determine this by
      * monitoring the committed offsets.
      *
-     * @param epoch         The controller epoch.
-     * @param batch         The batch of messages to write.
+     * @param epoch         the controller epoch
+     * @param batch         the batch of messages to write
      *
-     * @return              The offset of the message.
+     * @return              the offset of the last message in the batch
+     * @throws IllegalArgumentException if buffer allocatio failed and the client should backoff
      */
     long scheduleWrite(long epoch, List<ApiMessageAndVersion> batch);
 
+    /**
+     * Schedule a atomic write to the log.
+     *
+     * The write will be scheduled to happen at some time in the future.  All of the messages in batch
+     * will be appended atomically in one batch.  The listener may regard the write as successful
+     * if and only if the MetaLogManager reaches the given offset before renouncing its leadership.
+     * The listener should determine this by monitoring the committed offsets.
+     *
+     * @param epoch         the controller epoch
+     * @param batch         the batch of messages to write
+     *
+     * @return              the offset of the last message in the batch
+     * @throws IllegalArgumentException if buffer allocatio failed and the client should backoff
+     */
+    long scheduleAtomicWrite(long epoch, List<ApiMessageAndVersion> batch);
+
     /**
      * Renounce the leadership.
      *

metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java

@@ -135,18 +135,43 @@ public void testIncrementalAlterConfigs() {
         SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
         ConfigurationControlManager manager =
             new ConfigurationControlManager(new LogContext(), snapshotRegistry, CONFIGS);
-        assertEquals(new ControllerResult<Map<ConfigResource, ApiError>>(Collections.singletonList(
-            new ApiMessageAndVersion(new ConfigRecord().
-                setResourceType(TOPIC.id()).setResourceName("mytopic").
-                setName("abc").setValue("123"), (short) 0)),
-            toMap(entry(BROKER0, new ApiError(
-                Errors.INVALID_REQUEST, "A DELETE op was given with a non-null value.")),
-                entry(MYTOPIC, ApiError.NONE))),
-            manager.incrementalAlterConfigs(toMap(entry(BROKER0, toMap(
-                entry("foo.bar", entry(DELETE, "abc")),
-                entry("quux", entry(SET, "abc")))),
-            entry(MYTOPIC, toMap(
-                entry("abc", entry(APPEND, "123")))))));
+        assertEquals(
+            new ControllerResult<Map<ConfigResource, ApiError>>(
+                Collections.singletonList(
+                    new ApiMessageAndVersion(
+                        new ConfigRecord()
+                            .setResourceType(TOPIC.id())
+                            .setResourceName("mytopic")
+                            .setName("abc")
+                            .setValue("123"),
+                        (short) 0
+                    )
+                ),
+                toMap(
+                    entry(
+                        BROKER0,
+                        new ApiError(
+                            Errors.INVALID_REQUEST,
+                            "A DELETE op was given with a non-null value."
+                        )
+                    ),
+                    entry(MYTOPIC, ApiError.NONE)
+                ),
+                true
+            ),
+            manager.incrementalAlterConfigs(
+                toMap(
+                    entry(
+                        BROKER0,
+                        toMap(
+                            entry("foo.bar", entry(DELETE, "abc")),
+                            entry("quux", entry(SET, "abc"))
+                        )
+                    ),
+                    entry(MYTOPIC, toMap(entry("abc", entry(APPEND, "123"))))
+                )
+            )
+        );
     }
 
     @Test
@@ -184,20 +209,35 @@ public void testLegacyAlterConfigs() {
             new ApiMessageAndVersion(new ConfigRecord().
                 setResourceType(TOPIC.id()).setResourceName("mytopic").
                 setName("def").setValue("901"), (short) 0));
-        assertEquals(new ControllerResult<Map<ConfigResource, ApiError>>(
+        assertEquals(
+            new ControllerResult<Map<ConfigResource, ApiError>>(
                 expectedRecords1,
-                toMap(entry(MYTOPIC, ApiError.NONE))),
-            manager.legacyAlterConfigs(toMap(entry(MYTOPIC, toMap(
-                entry("abc", "456"), entry("def", "901"))))));
+                toMap(entry(MYTOPIC, ApiError.NONE)),
+                true
+            ),
+            manager.legacyAlterConfigs(
+                toMap(entry(MYTOPIC, toMap(entry("abc", "456"), entry("def", "901"))))
+            )
+        );
         for (ApiMessageAndVersion message : expectedRecords1) {
             manager.replay((ConfigRecord) message.message());
         }
-        assertEquals(new ControllerResult<Map<ConfigResource, ApiError>>(Arrays.asList(
-            new ApiMessageAndVersion(new ConfigRecord().
-                setResourceType(TOPIC.id()).setResourceName("mytopic").
-                setName("abc").setValue(null), (short) 0)),
-                toMap(entry(MYTOPIC, ApiError.NONE))),
-            manager.legacyAlterConfigs(toMap(entry(MYTOPIC, toMap(
-                entry("def", "901"))))));
+        assertEquals(
+            new ControllerResult<Map<ConfigResource, ApiError>>(
+                Arrays.asList(
+                    new ApiMessageAndVersion(
+                        new ConfigRecord()
+                            .setResourceType(TOPIC.id())
+                            .setResourceName("mytopic")
+                            .setName("abc")
+                            .setValue(null),
+                        (short) 0
+                    )
+                ),
+                toMap(entry(MYTOPIC, ApiError.NONE)),
+                true
+            ),
+            manager.legacyAlterConfigs(toMap(entry(MYTOPIC, toMap(entry("def", "901")))))
+        );
     }
 }