@Override
public void handleBatch(final List<ChangeEvent<Object, Object>> records,
                        final RecordCommitter<ChangeEvent<Object, Object>> committer)
        throws InterruptedException {
    final CountDownLatch latch = new CountDownLatch(records.size());
    for (ChangeEvent<Object, Object> record : records) {
        try {
            LOGGER.trace("Received event '{}'", record);
            Headers headers = convertKafkaHeaders(record);

            String topicName = streamNameMapper.map(record.destination());
            producer.send(new ProducerRecord<>(topicName, null, null, record.key(), record.value(), headers), (metadata, exception) -> {
                if (exception != null) {
                    LOGGER.error("Failed to send record to {}:", topicName, exception);
                    throw new DebeziumException(exception);
                }
                else {
                    LOGGER.trace("Sent message with offset: {}", metadata.offset());
                    latch.countDown();
                }
            });
            committer.markProcessed(record);
        }
        catch (Exception e) {
            throw new DebeziumException(e);
        }
    }

    latch.await();
    committer.markBatchFinished();
}