如何让Kafka MirrorMaker执行initTransactions以启用幂等性？

First, let's clarify the root cause of that IllegalStateException: When you set a transactional.id, Kafka requires the producer to explicitly call initTransactions() before sending any messages. The problem is that MirrorMaker 1 (MM1) doesn't have built-in logic to do this, while MirrorMaker 2 (MM2) (built on Kafka Connect) does—if configured correctly.

Let's break down the fix for each version:

For MirrorMaker 1 (MM1) #

MM1 is the older, standalone version of MirrorMaker, and it doesn't support transaction initialization out of the box. That's why setting transactional.id throws that error.

But here's the good news: You don't need transactions just to fix duplicate messages from network blips. Idempotence alone is enough to prevent duplicates, and MM1 supports idempotent producers without requiring transactional.id.

Follow these steps to fix the duplicate issue without transactions:

• Remove the transactional.id from your producer config entirely.
• Keep enable.idempotence=true (this is the key setting to stop duplicates).
• Pair it with these safe producer settings to maximize reliability:

  acks=all
  retries=3
  max.in.flight.requests.per.connection=1
  

Idempotent producers use a unique producer ID (PID) and sequence numbers to ensure that even if a message is retried due to network issues, it won't be duplicated in the target cluster.

If you absolutely need transactional semantics (not just idempotence), MM1 won't cut it—you'll need to upgrade to MM2.

For MirrorMaker 2 (MM2) #

MM2 is built on Kafka Connect, which added support for transactional producers in Kafka 2.5+. Connect automatically handles calling initTransactions() when you configure a transactional.id, so you just need to set up the right configs.

Here's how to configure it properly:

1. Ensure you're using Kafka 2.5 or newer—older versions don't support transactions in Connect.
2. Add these producer settings to your MM2 config file:

   # Enable idempotence and transactions
   producer.enable.idempotence=true
   producer.transactional.id=mm2-transaction-<your-mirror-group-name>
   producer.transaction.timeout.ms=60000
   
   # Configure transaction state log (required for Connect transactions)
   transaction.state.log.replication.factor=3
   transaction.state.log.min.isr=2
   transaction.state.log.num.partitions=50
   

   Replace <your-mirror-group-name> with a unique identifier for your MirrorMaker group to avoid conflicts.
3. Restart MM2—Kafka Connect will automatically initialize transactions on startup, so you won't see that initTransactions error anymore.

This setup will not only fix duplicates but also give you transactional guarantees if you need them (like atomic delivery of messages across topics).

内容的提问来源于stack exchange，提问作者gotner