public static ConnectorOffsets consumerGroupOffsetsToConnectorOffsets(Map<TopicPartition, OffsetAndMetadata> consumerGroupOffsets) {
        List<ConnectorOffset> connectorOffsets = new ArrayList<>();

        for (Map.Entry<TopicPartition, OffsetAndMetadata> topicPartitionOffset : consumerGroupOffsets.entrySet()) {
            Map<String, Object> partition = new HashMap<>();
            partition.put(KAFKA_TOPIC_KEY, topicPartitionOffset.getKey().topic());
            partition.put(KAFKA_PARTITION_KEY, topicPartitionOffset.getKey().partition());
            connectorOffsets.add(new ConnectorOffset(partition,
                    Collections.singletonMap(KAFKA_OFFSET_KEY, topicPartitionOffset.getValue().offset())));
        }

        return new ConnectorOffsets(connectorOffsets);
    }
@Test
    public void testConsumerGroupOffsetsToConnectorOffsets() {
        Map<TopicPartition, OffsetAndMetadata> consumerGroupOffsets = new HashMap<>();
        ConnectorOffsets connectorOffsets = SinkUtils.consumerGroupOffsetsToConnectorOffsets(consumerGroupOffsets);
        assertEquals(0, connectorOffsets.offsets().size());

        consumerGroupOffsets.put(new TopicPartition("test-topic", 0), new OffsetAndMetadata(100));

        connectorOffsets = SinkUtils.consumerGroupOffsetsToConnectorOffsets(consumerGroupOffsets);
        assertEquals(1, connectorOffsets.offsets().size());
        assertEquals(Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 100L), connectorOffsets.offsets().get(0).offset());

        Map<String, Object> expectedPartition = new HashMap<>();
        expectedPartition.put(SinkUtils.KAFKA_TOPIC_KEY, "test-topic");
        expectedPartition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
        assertEquals(expectedPartition, connectorOffsets.offsets().get(0).partition());
    }