@Override public FileSinkOperator.RecordWriter getHiveRecordWriter(JobConf jc,
      Path finalOutPath,
      Class<? extends Writable> valueClass,
      boolean isCompressed,
      Properties tableProperties,
      Progressable progress) {
    final String topic = jc.get(KafkaTableProperties.HIVE_KAFKA_TOPIC.getName());
    final Boolean optimisticCommit = jc.getBoolean(KafkaTableProperties.HIVE_KAFKA_OPTIMISTIC_COMMIT.getName(), false);
    final WriteSemantic
        writeSemantic =
        WriteSemantic.valueOf(jc.get(KafkaTableProperties.WRITE_SEMANTIC_PROPERTY.getName()));
    final Properties producerProperties = KafkaUtils.producerProperties(jc);
    final FileSinkOperator.RecordWriter recordWriter;
    switch (writeSemantic) {
    case AT_LEAST_ONCE:
      recordWriter = new SimpleKafkaWriter(topic, Utilities.getTaskId(jc), producerProperties);
      break;
    case EXACTLY_ONCE:
      FileSystem fs;
      try {
        fs = finalOutPath.getFileSystem(jc);
      } catch (IOException e) {
        LOG.error("Can not construct file system instance", e);
        throw new RuntimeException(e);
      }
      final String queryId = Preconditions.checkNotNull(jc.get(HiveConf.ConfVars.HIVE_QUERY_ID.varname, null));
      recordWriter =
          new TransactionalKafkaWriter(topic, producerProperties,
              new Path(Preconditions.checkNotNull(finalOutPath), queryId),
              fs,
              optimisticCommit);
      break;
    default:
      throw new IllegalArgumentException(String.format("Unknown delivery semantic [%s]", writeSemantic.toString()));
    }
    return recordWriter;
  }
Kafka Hive Output Format class used to write Hive Rows to a Kafka Queue.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaOutputFormat.java