KAFKA-14785: Connect offset read REST API

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

@@ -32,8 +32,6 @@
 import java.nio.file.NoSuchFileException;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -85,25 +83,7 @@ private void load() {
                 ByteBuffer key = (mapEntry.getKey() != null) ? ByteBuffer.wrap(mapEntry.getKey()) : null;
                 ByteBuffer value = (mapEntry.getValue() != null) ? ByteBuffer.wrap(mapEntry.getValue()) : null;
                 data.put(key, value);
-                if (key != null) {
-                    // The key should always be of the form [connectorName, partition] where connectorName is a
-                    // string value and partition is a Map<String, Object>
-                    try {
-                        // The topic parameter is irrelevant for the JsonConverter which is the internal converter used by
-                        // Connect workers.
-                        List<Object> keyValue = (List<Object>) keyConverter.toConnectData("", key.array()).value();
-                        String connectorName = (String) keyValue.get(0);
-                        Map<String, Object> partition = (Map<String, Object>) keyValue.get(1);
-                        connectorPartitions.computeIfAbsent(connectorName, ignored -> new HashSet<>());
-                        if (value == null) {
-                            connectorPartitions.get(connectorName).remove(partition);
-                        } else {
-                            connectorPartitions.get(connectorName).add(partition);
-                        }
-                    } catch (ClassCastException | IndexOutOfBoundsException e) {
-                        log.warn("Failed to deserialize offset key with an unexpected format", e);
-                    }
-                }
+                OffsetUtils.processPartitionKey(mapEntry.getKey(), mapEntry.getValue(), keyConverter, connectorPartitions);
             }
         } catch (NoSuchFileException | EOFException e) {
             // NoSuchFileException: Ignore, may be new.

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

@@ -79,6 +79,8 @@
 import java.util.concurrent.TimeoutException;
 import java.util.function.Supplier;
 
+import static org.apache.kafka.connect.util.ConnectUtils.className;
+
 /**
  * <p>
  * Provides persistent storage of Kafka Connect connector configurations in a Kafka topic.
@@ -1240,9 +1242,5 @@ else if (value instanceof Long)
         else
             throw new ConnectException("Expected integer value to be either Integer or Long");
     }
-
-    private String className(Object o) {
-        return o != null ? o.getClass().getName() : "null";
-    }
 }
 

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

@@ -48,8 +48,6 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
@@ -348,23 +346,7 @@ public Set<Map<String, Object>> connectorPartitions(String connectorName) {
             return;
         }
 
-        if (record.key() != null) {
-            // The key should always be a list of the form [connectorName, partition] where connectorName is a
-            // string value and partition is a Map<String, Object>
-            try {
-                List<Object> keyValue = (List<Object>) keyConverter.toConnectData(topic, record.key()).value();
-                String connectorName = (String) keyValue.get(0);
-                Map<String, Object> partition = (Map<String, Object>) keyValue.get(1);
-                connectorPartitions.computeIfAbsent(connectorName, ignored -> new HashSet<>());
-                if (record.value() == null) {
-                    connectorPartitions.get(connectorName).remove(partition);
-                } else {
-                    connectorPartitions.get(connectorName).add(partition);
-                }
-            } catch (ClassCastException | IndexOutOfBoundsException e) {
-                log.warn("Failed to deserialize offset key with an unexpected format", e);
-            }
-        }
+        OffsetUtils.processPartitionKey(record.key(), record.value(), keyConverter, connectorPartitions);
 
         ByteBuffer key = record.key() != null ? ByteBuffer.wrap(record.key()) : null;
 

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

@@ -19,10 +19,20 @@
 import org.apache.kafka.connect.data.ConnectSchema;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.errors.DataException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
+import java.util.Set;
+
+import static org.apache.kafka.connect.util.ConnectUtils.className;
 
 public class OffsetUtils {
+
+    private static final Logger log = LoggerFactory.getLogger(OffsetUtils.class);
+
     @SuppressWarnings("unchecked")
     public static void validateFormat(Object offsetData) {
         if (offsetData == null)
@@ -53,4 +63,64 @@ public static <K, V> void validateFormat(Map<K, V> offsetData) {
                 throw new DataException("Offsets may only contain primitive types as values, but field " + entry.getKey() + " contains " + schemaType);
         }
     }
+
+    /**
+     * Parses a partition key that is read back from an offset backing store and add / remove the partition in the
+     * provided {@code connectorPartitions} map. If the partition key has an unexpected format, a warning log is emitted
+     * and nothing is added / removed in the {@code connectorPartitions} map.
+     * @param partitionKey the partition key to be processed
+     * @param offsetValue the offset value corresponding to the partition key; determines whether the partition should
+     *                    be added to the {@code connectorPartitions} map or removed depending on whether the offset
+     *                    value is null or not.
+     * @param keyConverter the key converter to deserialize the partition key
+     * @param connectorPartitions the map from connector names to its set of partitions which needs to be updated after
+     *                            processing
+     */
+    @SuppressWarnings("unchecked")
+    public static void processPartitionKey(byte[] partitionKey, byte[] offsetValue, Converter keyConverter,
+                                           Map<String, Set<Map<String, Object>>> connectorPartitions) {
+
+        // The key is expected to always be of the form [connectorName, partition] where connectorName is a
+        // string value and partition is a Map<String, Object>
+
+        if (partitionKey == null) {
+            log.warn("Ignoring offset partition key with an unexpected null value");
+            return;
+        }
+        // The topic parameter is irrelevant for the JsonConverter which is the internal converter used by
+        // Connect workers.
+        Object deserializedValue = keyConverter.toConnectData("", partitionKey).value();
+        if (!(deserializedValue instanceof List)) {
+            log.warn("Ignoring offset partition key with an unexpected format. Expected type: {}, actual type: {}",
+                    List.class.getName(), className(deserializedValue));
+            return;
+        }
+
+        List<Object> keyList = (List<Object>) deserializedValue;
+        if (keyList.size() != 2) {
+            log.warn("Ignoring offset partition key with an unexpected number of elements. Expected: 2, actual: {}", keyList.size());
+            return;
+        }
+
+        if (!(keyList.get(0) instanceof String)) {
+            log.warn("Ignoring offset partition key with an unexpected format for the first element in the partition key list. " +
+                    "Expected type: {}, actual type: {}", String.class.getName(), className(keyList.get(0)));
+            return;
+        }
+
+        if (!(keyList.get(1) instanceof Map)) {
+            log.warn("Ignoring offset partition key with an unexpected format for the second element in the partition key list. " +
+                    "Expected type: {}, actual type: {}", Map.class.getName(), className(keyList.get(1)));
+            return;
+        }
+
+        String connectorName = (String) keyList.get(0);
+        Map<String, Object> partition = (Map<String, Object>) keyList.get(1);
+        connectorPartitions.computeIfAbsent(connectorName, ignored -> new HashSet<>());
+        if (offsetValue == null) {
+            connectorPartitions.get(connectorName).remove(partition);
+        } else {
+            connectorPartitions.get(connectorName).add(partition);
+        }
+    }
 }

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

@@ -209,4 +209,13 @@ public static String clientIdBase(WorkerConfig config) {
         }
         return result + "-";
     }
+
+    /**
+     * Get the class name for an object in a null-safe manner.
+     * @param o the object whose class name is to be returned
+     * @return "null" if the object is null; or else the object's class name
+     */
+    public static String className(Object o) {
+        return o != null ? o.getClass().getName() : "null";
+    }
 }

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

@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.storage;
+
+import org.apache.kafka.common.utils.LogCaptureAppender;
+import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.json.JsonConverter;
+import org.apache.kafka.connect.json.JsonConverterConfig;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+public class OffsetUtilsTest {
+
+    private static final JsonConverter CONVERTER = new JsonConverter();
+
+    static {
+        CONVERTER.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false"), true);
+    }
+
+    @Test
+    public void testValidateFormatNotMap() {
+        DataException e = assertThrows(DataException.class, () -> OffsetUtils.validateFormat(new Object()));
+        assertThat(e.getMessage(), containsString("Offsets must be specified as a Map"));
+    }
+
+    @Test
+    public void testValidateFormatMapWithNonStringKeys() {
+        Map<Object, Object> offsetData = new HashMap<>();
+        offsetData.put("k1", "v1");
+        offsetData.put(1, "v2");
+        DataException e = assertThrows(DataException.class, () -> OffsetUtils.validateFormat(offsetData));
+        assertThat(e.getMessage(), containsString("Offsets may only use String keys"));
+    }
+
+    @Test
+    public void testValidateFormatMapWithNonPrimitiveKeys() {
+        Map<Object, Object> offsetData = Collections.singletonMap("key", new Object());
+        DataException e = assertThrows(DataException.class, () -> OffsetUtils.validateFormat(offsetData));
+        assertThat(e.getMessage(), containsString("Offsets may only contain primitive types as values"));
+
+        Map<Object, Object> offsetData2 = Collections.singletonMap("key", new ArrayList<>());
+        e = assertThrows(DataException.class, () -> OffsetUtils.validateFormat(offsetData2));
+        assertThat(e.getMessage(), containsString("Offsets may only contain primitive types as values"));
+    }
+
+    @Test
+    public void testValidateFormatWithValidFormat() {
+        Map<Object, Object> offsetData = Collections.singletonMap("key", 1);
+        // Expect no exception to be thrown
+        OffsetUtils.validateFormat(offsetData);
+    }
+
+    @Test
+    public void testProcessPartitionKeyNotList() {
+        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(OffsetUtils.class)) {
+            Map<String, Set<Map<String, Object>>> connectorPartitions = new HashMap<>();
+            OffsetUtils.processPartitionKey(serializePartitionKey(new HashMap<>()), new byte[0], CONVERTER, connectorPartitions);
+            // Expect no partition to be added to the map since the partition key is of an invalid format
+            assertEquals(0, connectorPartitions.size());
+            assertEquals(1, logCaptureAppender.getMessages().size());
+            assertThat(logCaptureAppender.getMessages().get(0),
+                    containsString("Ignoring offset partition key with an unexpected format"));
+        }
+    }
+
+    @Test
+    public void testProcessPartitionKeyListWithOneElement() {
+        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(OffsetUtils.class)) {
+            Map<String, Set<Map<String, Object>>> connectorPartitions = new HashMap<>();
+            OffsetUtils.processPartitionKey(serializePartitionKey(Collections.singletonList("")), new byte[0], CONVERTER, connectorPartitions);
+            // Expect no partition to be added to the map since the partition key is of an invalid format
+            assertEquals(0, connectorPartitions.size());
+            assertEquals(1, logCaptureAppender.getMessages().size());
+            assertThat(logCaptureAppender.getMessages().get(0),
+                    containsString("Ignoring offset partition key with an unexpected number of elements"));
+        }
+    }
+
+    @Test
+    public void testProcessPartitionKeyListWithElementsOfWrongType() {
+        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(OffsetUtils.class)) {
+            Map<String, Set<Map<String, Object>>> connectorPartitions = new HashMap<>();
+            OffsetUtils.processPartitionKey(serializePartitionKey(Arrays.asList(1, new HashMap<>())), new byte[0], CONVERTER, connectorPartitions);
+            // Expect no partition to be added to the map since the partition key is of an invalid format
+            assertEquals(0, connectorPartitions.size());
+            assertEquals(1, logCaptureAppender.getMessages().size());
+            assertThat(logCaptureAppender.getMessages().get(0),
+                    containsString("Ignoring offset partition key with an unexpected format for the first element in the partition key list"));
+
+            OffsetUtils.processPartitionKey(serializePartitionKey(Arrays.asList("connector-name", new ArrayList<>())), new byte[0], CONVERTER, connectorPartitions);
+            // Expect no partition to be added to the map since the partition key is of an invalid format
+            assertEquals(0, connectorPartitions.size());
+            assertEquals(2, logCaptureAppender.getMessages().size());
+            assertThat(logCaptureAppender.getMessages().get(1),
+                    containsString("Ignoring offset partition key with an unexpected format for the second element in the partition key list"));
+        }
+    }
+
+    @Test
+    public void testProcessPartitionKeyValidList() {
+        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(OffsetUtils.class)) {
+            Map<String, Set<Map<String, Object>>> connectorPartitions = new HashMap<>();
+            OffsetUtils.processPartitionKey(serializePartitionKey(Arrays.asList("connector-name", new HashMap<>())), new byte[0], CONVERTER, connectorPartitions);
+            assertEquals(1, connectorPartitions.size());
+            assertEquals(0, logCaptureAppender.getMessages().size());
+        }
+    }
+
+    private byte[] serializePartitionKey(Object key) {
+        return CONVERTER.fromConnectData("", null, key);
+    }
+}