class KafkaSinkTask extends SinkTask {

    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
        LOGGER.debug("Flushing kafka sink");

        try {
            producer.flush();
        } catch (IOException e) {
            LOGGER.debug("IOException on flush, re-throwing as retriable", e);
            // Re-throw exception as connect retriable since we just want connect to keep retrying forever
            throw new RetriableException(e);
        }

        super.flush(offsets);
    }

}