必要ライブラリ

slf4j-simple-1.7.36

事前準備

Kafka brokcerの起動
下記のコマンドで立ち上げを行います。
/bin/kafka-server-start.sh /config/server.properties

Coding

Kafkaのプロパティを定義します。
final Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
BOOTSTRAP_SERVERS);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
“my-transactional-id”);

ProducerConfig.BOOTSTRAP_SERVERS_CONFIGには接続するKafkaサーバの[URL:PORT]を記載します。
ProducerConfig.TRANSACTIONAL_ID_CONFIGにはトランザクションのIDを記載します。今回は任意の値で使います。

Producerを定義し、トランザクションを初期化します。
Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
producer.initTransactions();

トランザクション処理を開始します。
producer.beginTransaction();

Kafkaへイベントを送信します。
producer.send(new ProducerRecord<>(TOPIC, Integer.toString(i), “Java Producer Pub”));

今回のTOPIC、値については任意で設定し、行います。
5. 4の処理が問題なく、終わったらコミット、接続をクローズします。

	 producer.commitTransaction();
	 producer.close();

参考

下記のレポジトリにソースコードを確認いただけます。
https://github.com/dlstjq7685/KafkaSimpleProducerTuttorial

ソースコードはkafkaのJavaDocの内容に沿って作成しております。
あくまで参考程度で書いたものですので、
その他の設定については別記事を参考にしてください。