KafkaDestination extends AbstractDestination<T> { public KafkaDestination( final String prefix, final KafkaProducerConfiguration producerConfig, final BatchMapper<Mutation<?>, T> mapper, final DestinationMetrics metrics, final long delaySendMs) { super(mapper, metrics, delaySendMs); topicNamePrefix = Optional.ofNullable(prefix).orElse(DEFAULT_TOPIC_PREFIX); Properties props = new Properties(); setKafkaDefaultConfigs(props, producerConfig.getBootstrapServers()); kafkaProducer = new KafkaProducer<>(props); } KafkaDestination(
      final String prefix,
      final KafkaProducerConfiguration producerConfig,
      final BatchMapper<Mutation<?>, T> mapper,
      final DestinationMetrics metrics,
      final long delaySendMs); @Override void publish(List<T> messages);  }
@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(); }