Kafka Transaktions Produzent - read_committed zeigt die Datensätze trotz Abbruch

Question

ich dieses einfache Programm schrieb den neuen Transaktions Produzent in Kafka zu testen:

package test; 

import org.apache.kafka.clients.producer.KafkaProducer; 
import org.apache.kafka.clients.producer.ProducerConfig; 
import org.apache.kafka.clients.producer.ProducerRecord; 

import java.util.Properties; 


class kafkatest { 

    public static void main(String[] args) { 

     Properties props = new Properties(); 
     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); 
     props.put(ProducerConfig.CLIENT_ID_CONFIG, "hello-world-producer"); 
     props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); 
     props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test"); 
     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); 

     KafkaProducer producer = new KafkaProducer(props); 

     producer.initTransactions(); 
     producer.beginTransaction(); 

     producer.send(new ProducerRecord<>("topic", "hello", "world")); 
     producer.flush(); 

     producer.abortTransaction(); 

     producer.close(); 
    } 
} 

aber wenn ich mit isolation.level=read_committed verbrauchen, dass sehr Datensatz zeigt sich:

--- ~ » kafka-console-consumer --bootstrap-server localhost:9092 \ 
     --topic topic \ 
     --from-beginning \ 
     --consumer-property isolation.level=read_committed 

world 

Was fehlt mir?

Answer 1

Um read_committed mit der Konsole Verbraucher zu verwenden, müssen Sie die --isolation-level Option angeben:

kafka-console-consumer.sh --bootstrap-server localhost:9092 \ 
--topic topic --from-beginning --isolation-level=read_committed 

Diese Option sonst standardmäßig read_uncommitted und überschreibt den Wert, den Sie über --consumer-property vorbei sind.