public static Map<TopicPartition, Long> parseSinkConnectorOffsets(Map<Map<String, ?>, Map<String, ?>> partitionOffsets) {
        Map<TopicPartition, Long> parsedOffsetMap = new HashMap<>();

        for (Map.Entry<Map<String, ?>, Map<String, ?>> partitionOffset : partitionOffsets.entrySet()) {
            Map<String, ?> partitionMap = partitionOffset.getKey();
            if (partitionMap == null) {
                throw new BadRequestException("The partition for a sink connector offset cannot be null or missing");
            }
            if (!partitionMap.containsKey(KAFKA_TOPIC_KEY) || !partitionMap.containsKey(KAFKA_PARTITION_KEY)) {
                throw new BadRequestException(String.format("The partition for a sink connector offset must contain the keys '%s' and '%s'",
                        KAFKA_TOPIC_KEY, KAFKA_PARTITION_KEY));
            }
            if (partitionMap.get(KAFKA_TOPIC_KEY) == null) {
                throw new BadRequestException("Kafka topic names must be valid strings and may not be null");
            }
            if (partitionMap.get(KAFKA_PARTITION_KEY) == null) {
                throw new BadRequestException("Kafka partitions must be valid numbers and may not be null");
            }
            String topic = String.valueOf(partitionMap.get(KAFKA_TOPIC_KEY));
            int partition;
            try {
                // We parse it this way because both "10" and 10 should be accepted as valid partition values in the REST API's
                // JSON request payload. If it throws an exception, we should propagate it since it's indicative of a badly formatted value.
                partition = Integer.parseInt(String.valueOf(partitionMap.get(KAFKA_PARTITION_KEY)));
            } catch (Exception e) {
                throw new BadRequestException("Failed to parse the following Kafka partition value in the provided offsets: '" +
                        partitionMap.get(KAFKA_PARTITION_KEY) + "'. Partition values for sink connectors need " +
                        "to be integers.", e);
            }
            TopicPartition tp = new TopicPartition(topic, partition);

            Map<String, ?> offsetMap = partitionOffset.getValue();

            if (offsetMap == null) {
                // represents an offset reset
                parsedOffsetMap.put(tp, null);
            } else {
                if (!offsetMap.containsKey(KAFKA_OFFSET_KEY)) {
                    throw new BadRequestException(String.format("The offset for a sink connector should either be null or contain " +
                            "the key '%s'", KAFKA_OFFSET_KEY));
                }
                long offset;
                try {
                    // We parse it this way because both "1000" and 1000 should be accepted as valid offset values in the REST API's
                    // JSON request payload. If it throws an exception, we should propagate it since it's indicative of a badly formatted value.
                    offset = Long.parseLong(String.valueOf(offsetMap.get(KAFKA_OFFSET_KEY)));
                } catch (Exception e) {
                    throw new BadRequestException("Failed to parse the following Kafka offset value in the provided offsets: '" +
                            offsetMap.get(KAFKA_OFFSET_KEY) + "'. Offset values for sink connectors need " +
                            "to be integers.", e);
                }
                parsedOffsetMap.put(tp, offset);
            }
        }

        return parsedOffsetMap;
    }
@Test
    public void testNullOffset() {
        Map<String, Object> partitionMap = new HashMap<>();
        partitionMap.put(SinkUtils.KAFKA_TOPIC_KEY, "topic");
        partitionMap.put(SinkUtils.KAFKA_PARTITION_KEY, 10);
        Map<Map<String, ?>, Map<String, ?>> partitionOffsets = new HashMap<>();
        partitionOffsets.put(partitionMap, null);
        Map<TopicPartition, Long> parsedOffsets = SinkUtils.parseSinkConnectorOffsets(partitionOffsets);
        assertEquals(1, parsedOffsets.size());
        assertNull(parsedOffsets.values().iterator().next());
    }