Apache Kafka - Producer Pause auf initTransaction

Question

Ich teste einen elementaren Transaktionsfall, aber es bleibt auf producer.initTransaction(); Kann irgendeine Konfiguration falsch sein?

public static void main(String[] argv) throws Exception { 
    String topicName = "helloworldpartitioned"; 
    // Configure the Producer 
    Properties configProperties = new Properties(); 
    configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); 
    configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); 
    configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); 
    configProperties.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "1000"); 
    configProperties.put(ProducerConfig.CLIENT_ID_CONFIG, "test"); 
//  configProperties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, Boolean.TRUE); 
    configProperties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, PRODUCER_TRANSACTIONAL_ID); 

    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configProperties); 
    producer.initTransactions(); 
    producer.beginTransaction(); 
     String line = "TestMessage"; 
     System.out.println("Inizializzo la transazione"); 
     ProducerRecord<String, String> rec = new ProducerRecord<String, String>(topicName, line); 
     producer.send(rec); 
    // Thread.sleep(5000); 
    producer.commitTransaction(); 
    producer.flush(); 
    producer.close(); 
} 

Answer 1

Versuchen Sie:

producer.initTransactions(); 
producer.beginTransaction(); 
    String line = "TestMessage"; 
    System.out.println("Inizializzo la transazione"); 
    ProducerRecord<String, String> rec = new ProducerRecord<String, String>(topicName, line); 
    producer.send(rec); 


// Thread.sleep(5000); 
producer.commitTransaction(); 
producer.flush(); 
producer.close();