### Question:
KafkaSinkTask extends SinkTask { @Override public void put(Collection<SinkRecord> collection) { for(SinkRecord sinkRecord : collection) { if (sinkRecord.keySchema() != Schema.OPTIONAL_BYTES_SCHEMA || sinkRecord.valueSchema() != Schema.OPTIONAL_BYTES_SCHEMA) throw new IllegalStateException("Expected sink record key/value to be optional bytes, but saw instead key: " + sinkRecord.keySchema() + " value: " + sinkRecord.valueSchema() + ". Must use converter: " + "org.apache.kafka.connect.converters.ByteArrayConverter"); LOGGER.debug("Sending record {}", sinkRecord); try { producer.send(new ProducerRecord<>(sinkRecord.topic(), sinkRecord.kafkaPartition(), (byte[]) sinkRecord.key(), (byte[]) sinkRecord.value())); } catch (KafkaException e) { throw new RetriableException(e); } } }  @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 put() { task.put(Collections.singletonList(sinkRecord)); verify(kafkaProducer).send(anyObject()); }
@Test (expected = IllegalStateException.class) public void put_recordKeyIsNotNullOrBytes() { sinkRecord = new SinkRecord("topic", 0, Schema.STRING_SCHEMA, "key", Schema.OPTIONAL_BYTES_SCHEMA, "value".getBytes(), 0L); task.put(Collections.singletonList(sinkRecord)); }
@Test (expected = IllegalStateException.class) public void put_recordValueIsNotNullOrBytes() { sinkRecord = new SinkRecord("topic", 0, Schema.OPTIONAL_BYTES_SCHEMA, "key".getBytes(), Schema.STRING_SCHEMA, "value", 0L); task.put(Collections.singletonList(sinkRecord)); }
@Test (expected = RetriableException.class) public void put_producerThrowsException() { when(kafkaProducer.send(anyObject())).thenThrow(new KafkaException()); task.put(Collections.singletonList(sinkRecord)); }