@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 deserializedKey;
        try {
            deserializedKey = keyConverter.toConnectData("", partitionKey).value();
        } catch (DataException e) {
            log.warn("Ignoring offset partition key with unknown serialization. Expected json.", e);
            return;
        }
        if (!(deserializedKey instanceof List)) {
            log.warn("Ignoring offset partition key with an unexpected format. Expected type: {}, actual type: {}",
                    List.class.getName(), className(deserializedKey));
            return;
        }

        List<Object> keyList = (List<Object>) deserializedKey;
        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)) {
            if (keyList.get(1) != null) {
                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);
        }
    }
@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());
        }
    }