### Question:
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); throw new RetriableException(e); } super.flush(offsets); }  @Override String version(); @Override void start(Map<String, String> taskConfig); @Override void put(Collection<SinkRecord> collection); @Override void flush(Map<TopicPartition, OffsetAndMetadata> offsets); @Override void stop();  }### Answer:
@Test public void flush() { task.put(Collections.singletonList(sinkRecord)); task.flush(Collections.emptyMap()); verify(kafkaProducer).flush(); }