Merge branch '3.4' of https://github.com/apache/kafka into 3.4

Author: showuon

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTask.java

@@ -233,6 +233,7 @@ protected void finalOffsetCommit(boolean failed) {
     @Override
     public void removeMetrics() {
         Utils.closeQuietly(transactionMetrics, "source task transaction metrics tracker");
+        super.removeMetrics();
     }
 
     @Override

connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java

@@ -637,7 +637,7 @@ public void putTargetState(String connector, TargetState state) {
         byte[] serializedTargetState = converter.fromConnectData(topic, TARGET_STATE_V0, connectTargetState);
         log.debug("Writing target state {} for connector {}", state, connector);
         try {
-            configLog.send(TARGET_STATE_KEY(connector), serializedTargetState).get(READ_WRITE_TOTAL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+            configLog.sendWithReceipt(TARGET_STATE_KEY(connector), serializedTargetState).get(READ_WRITE_TOTAL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
         } catch (InterruptedException | ExecutionException | TimeoutException e) {
             log.error("Failed to write target state to Kafka", e);
             throw new ConnectException("Error writing target state to Kafka", e);
@@ -789,7 +789,7 @@ private void sendPrivileged(List<ProducerKeyValue> keyValues, Timer timer) throw
         if (!usesFencableWriter) {
             List<Future<RecordMetadata>> producerFutures = new ArrayList<>();
             keyValues.forEach(
-                    keyValue -> producerFutures.add(configLog.send(keyValue.key, keyValue.value))
+                    keyValue -> producerFutures.add(configLog.sendWithReceipt(keyValue.key, keyValue.value))
             );
 
             timer.update();

connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java

@@ -77,6 +77,10 @@
  *     calling class keeps track of state based on the log and only writes to it when consume callbacks are invoked
  *     and only reads it in {@link #readToEnd(Callback)} callbacks then no additional synchronization will be required.
  * </p>
+ * <p>
+ *     This is a useful utility that has been used outside of Connect. This isn't in Connect's public API,
+ *     but we've tried to maintain the method signatures and backward compatibility since early Kafka versions.
+ * </p>
  */
 public class KafkaBasedLog<K, V> {
     private static final Logger log = LoggerFactory.getLogger(KafkaBasedLog.class);
@@ -351,6 +355,31 @@ public Future<Void> readToEnd() {
         return future;
     }
 
+    /**
+     * Send a record asynchronously to the configured {@link #topic} without using a producer callback.
+     * <p>
+     * This method exists for backward compatibility reasons and delegates to the newer
+     * {@link #sendWithReceipt(Object, Object)} method that returns a future.
+     * @param key the key for the {@link ProducerRecord}
+     * @param value the value for the {@link ProducerRecord}
+     */
+    public void send(K key, V value) {
+        sendWithReceipt(key, value);
+    }
+
+    /**
+     * Send a record asynchronously to the configured {@link #topic}.
+     * <p>
+     * This method exists for backward compatibility reasons and delegates to the newer
+     * {@link #sendWithReceipt(Object, Object, org.apache.kafka.clients.producer.Callback)} method that returns a future.
+     * @param key the key for the {@link ProducerRecord}
+     * @param value the value for the {@link ProducerRecord}
+     * @param callback the callback to invoke after completion; can be null if no callback is desired
+     */
+    public void send(K key, V value, org.apache.kafka.clients.producer.Callback callback) {
+        sendWithReceipt(key, value, callback);
+    }
+
     /**
      * Send a record asynchronously to the configured {@link #topic} without using a producer callback.
      * @param key the key for the {@link ProducerRecord}
@@ -359,20 +388,20 @@ public Future<Void> readToEnd() {
      * @return the future from the call to {@link Producer#send}. {@link Future#get} can be called on this returned
      *         future if synchronous behavior is desired.
      */
-    public Future<RecordMetadata> send(K key, V value) {
-        return send(key, value, null);
+    public Future<RecordMetadata> sendWithReceipt(K key, V value) {
+        return sendWithReceipt(key, value, null);
     }
 
     /**
-     * Send a record asynchronously to the configured {@link #topic}
+     * Send a record asynchronously to the configured {@link #topic}.
      * @param key the key for the {@link ProducerRecord}
      * @param value the value for the {@link ProducerRecord}
      * @param callback the callback to invoke after completion; can be null if no callback is desired
      *
      * @return the future from the call to {@link Producer#send}. {@link Future#get} can be called on this returned
      *         future if synchronous behavior is desired.
      */
-    public Future<RecordMetadata> send(K key, V value, org.apache.kafka.clients.producer.Callback callback) {
+    public Future<RecordMetadata> sendWithReceipt(K key, V value, org.apache.kafka.clients.producer.Callback callback) {
         return producer.orElseThrow(() ->
                 new IllegalStateException("This KafkaBasedLog was created in read-only mode and does not support write operations")
         ).send(new ProducerRecord<>(topic, key, value), callback);

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ExactlyOnceWorkerSourceTaskTest.java

@@ -19,6 +19,7 @@
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.RecordTooLargeException;
@@ -82,7 +83,9 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 
+import static java.util.Collections.emptySet;
 import static org.apache.kafka.connect.integration.MonitorableSourceConnector.TOPIC_CONFIG;
 import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG;
 import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG;
@@ -291,6 +294,23 @@ private void createWorkerTask(TargetState initialState, Converter keyConverter,
                 sourceConfig, Runnable::run, preProducerCheck, postProducerCheck);
     }
 
+    @Test
+    public void testRemoveMetrics() {
+        createWorkerTask();
+
+        workerTask.removeMetrics();
+
+        assertEquals(emptySet(), filterToTaskMetrics(metrics.metrics().metrics().keySet()));
+    }
+
+    private Set<MetricName> filterToTaskMetrics(Set<MetricName> metricNames) {
+        return metricNames
+                .stream()
+                .filter(m -> metrics.registry().taskGroupName().equals(m.group())
+                        || metrics.registry().sourceTaskGroupName().equals(m.group()))
+                .collect(Collectors.toSet());
+    }
+
     @Test
     public void testStartPaused() throws Exception {
         createWorkerTask(TargetState.PAUSED);

connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaConfigBackingStoreTest.java

@@ -344,7 +344,7 @@ public void testPutConnectorConfigProducerError() throws Exception {
 
         expectConvert(KafkaConfigBackingStore.CONNECTOR_CONFIGURATION_V0, CONNECTOR_CONFIG_STRUCTS.get(0), CONFIGS_SERIALIZED.get(0));
 
-        storeLog.send(EasyMock.anyObject(), EasyMock.anyObject());
+        storeLog.sendWithReceipt(EasyMock.anyObject(), EasyMock.anyObject());
         EasyMock.expectLastCall().andReturn(producerFuture);
 
         producerFuture.get(EasyMock.anyLong(), EasyMock.anyObject());
@@ -379,13 +379,13 @@ public void testRemoveConnectorConfigSlowProducer() throws Exception {
         @SuppressWarnings("unchecked")
         Future<RecordMetadata> connectorConfigProducerFuture = PowerMock.createMock(Future.class);
         // tombstone for the connector config
-        storeLog.send(EasyMock.anyObject(), EasyMock.isNull());
+        storeLog.sendWithReceipt(EasyMock.anyObject(), EasyMock.isNull());
         EasyMock.expectLastCall().andReturn(connectorConfigProducerFuture);
 
         @SuppressWarnings("unchecked")
         Future<RecordMetadata> targetStateProducerFuture = PowerMock.createMock(Future.class);
         // tombstone for the connector target state
-        storeLog.send(EasyMock.anyObject(), EasyMock.isNull());
+        storeLog.sendWithReceipt(EasyMock.anyObject(), EasyMock.isNull());
         EasyMock.expectLastCall().andReturn(targetStateProducerFuture);
 
         connectorConfigProducerFuture.get(EasyMock.eq(READ_WRITE_TOTAL_TIMEOUT_MS), EasyMock.anyObject());
@@ -460,7 +460,7 @@ public void testWritePrivileges() throws Exception {
 
         // In the meantime, write a target state (which doesn't require write privileges)
         expectConvert(KafkaConfigBackingStore.TARGET_STATE_V0, TARGET_STATE_PAUSED, CONFIGS_SERIALIZED.get(1));
-        storeLog.send("target-state-" + CONNECTOR_IDS.get(1), CONFIGS_SERIALIZED.get(1));
+        storeLog.sendWithReceipt("target-state-" + CONNECTOR_IDS.get(1), CONFIGS_SERIALIZED.get(1));
         EasyMock.expectLastCall().andReturn(producerFuture);
         producerFuture.get(EasyMock.anyLong(), EasyMock.anyObject());
         EasyMock.expectLastCall().andReturn(null);
@@ -1601,7 +1601,7 @@ private void expectConvertWriteRead(final String configKey, final Schema valueSc
             EasyMock.expect(converter.fromConnectData(EasyMock.eq(TOPIC), EasyMock.eq(valueSchema), EasyMock.capture(capturedRecord)))
                     .andReturn(serialized);
 
-        storeLog.send(EasyMock.eq(configKey), EasyMock.aryEq(serialized));
+        storeLog.sendWithReceipt(EasyMock.eq(configKey), EasyMock.aryEq(serialized));
         EasyMock.expectLastCall().andReturn(producerFuture);
 
         producerFuture.get(EasyMock.anyLong(), EasyMock.anyObject());