@SuppressWarnings("unchecked")
@Test
public void KafkaDestination() throws Exception {

  createKafkaTopic(TOPIC);
  KafkaProducerConfiguration configs = new KafkaProducerConfiguration(this.bootstrapServers());
  KafkaDestination kafkaDestination = new KafkaDestination(null, configs, x -> x, metrics, 0L);
  List<Mutation> messages = new ArrayList<>();
  messages.add(createMutation(MutationType.INSERT));
  messages.add(createMutation(MutationType.UPDATE));
  messages.add(createMutation(MutationType.DELETE));
  kafkaDestination.publish(messages);
  Properties props = new Properties();
  props.setProperty("bootstrap.servers", this.bootstrapServers());
  props.setProperty(
      "key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
  props.setProperty(
      "value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
  KafkaConsumer<byte[], byte[]> kafkaConsumer = new KafkaConsumer<>(props);
  kafkaConsumer.assign(Collections.singletonList(new TopicPartition(TOPIC, 0)));
  kafkaConsumer.seekToBeginning(new TopicPartition(TOPIC, 0));
  List<ConsumerRecords<byte[], byte[]>> records = new ArrayList<>();
  ConsumerRecords<byte[], byte[]> record;
  long startMs = current();
  while (current() - startMs <= 10000L) {
    record = kafkaConsumer.poll(1000L);
    records.add(record);
    if (records.size() == 3) break;
  }
  Assert.assertEquals(records.size(), 3);
  for (ConsumerRecords<byte[], byte[]> consumerRecords : records) {
    for (ConsumerRecord<byte[], byte[]> consumerRecord : consumerRecords) {
      com.airbnb.jitney.event.spinaltap.v1.Mutation mutation =
          getMutation(consumerRecord.value());
      switch (mutation.getType()) {
        case INSERT:
          Assert.assertEquals(mutation, createMutation(MutationType.INSERT));
          break;
        case UPDATE:
          Assert.assertEquals(mutation, createMutation(MutationType.UPDATE));
          break;
        case DELETE:
          Assert.assertEquals(mutation, createMutation(MutationType.DELETE));
          break;
      }
    }
  }
  kafkaDestination.close();
  kafkaConsumer.close();
}