@Test public void testSend() throws IOException {
    String topic = UUID.randomUUID().toString();
    Properties properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaBrokerResource.BROKER_IP_PORT);
    SimpleKafkaWriter
        writer =
        new SimpleKafkaWriter(topic,
            null, properties);
    RECORDS_WRITABLES.forEach(kafkaRecordWritable -> {
      try {
        writer.write(kafkaRecordWritable);
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    });
    writer.close(false);
    Assert.assertEquals(RECORD_NUMBER, writer.getSentRecords());
    Assert.assertEquals(0, writer.getLostRecords());
    Set<TopicPartition> assignment = Collections.singleton(new TopicPartition(topic, 0));
    consumer.assign(assignment);
    consumer.seekToBeginning(assignment);
    long numRecords = 0;
    while (numRecords < RECORD_NUMBER) {
      ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(1000));

      Assert.assertFalse(records.records(new TopicPartition(topic, 0))
          .stream()
          .anyMatch(consumerRecord -> !RECORDS_WRITABLES.contains(new KafkaWritable(0,
              consumerRecord.timestamp(),
              consumerRecord.value(),
              consumerRecord.key()))));

      numRecords += records.count();
    }
    Assert.assertEquals(RECORD_NUMBER, numRecords);
  }
Test class for Kafka simple writer.
kafka-handler/src/test/org/apache/hadoop/hive/kafka/SimpleKafkaWriterTest.java