@Override
    public DeleteRecordsRequest.Builder buildBatchedRequest(int brokerId, Set<TopicPartition> keys) {
        Map<String, DeleteRecordsRequestData.DeleteRecordsTopic> deletionsForTopic = new HashMap<>();
        for (TopicPartition topicPartition : keys) {
            RecordsToDelete toDelete = recordsToDelete.get(topicPartition);
            DeleteRecordsRequestData.DeleteRecordsTopic deleteRecords = deletionsForTopic.computeIfAbsent(
                    topicPartition.topic(),
                    key -> new DeleteRecordsRequestData.DeleteRecordsTopic().setName(topicPartition.topic())
            );
            deleteRecords.partitions().add(new DeleteRecordsRequestData.DeleteRecordsPartition()
                    .setPartitionIndex(topicPartition.partition())
                    .setOffset(toDelete.beforeOffset()));
        }

        DeleteRecordsRequestData data = new DeleteRecordsRequestData()
                .setTopics(new ArrayList<>(deletionsForTopic.values()))
                .setTimeoutMs(timeout);
        return new DeleteRecordsRequest.Builder(data);
    }
@Test
    public void testBuildRequestSimple() {
        DeleteRecordsHandler handler = new DeleteRecordsHandler(recordsToDelete, logContext, timeout);
        DeleteRecordsRequest request = handler.buildBatchedRequest(node1.id(), mkSet(t0p0, t0p1)).build();
        List<DeleteRecordsRequestData.DeleteRecordsTopic> topics = request.data().topics();
        assertEquals(1, topics.size());
        DeleteRecordsRequestData.DeleteRecordsTopic topic = topics.get(0);
        assertEquals(2, topic.partitions().size());
    }