Kafka Streams Exactly-Once v2 事务处理与状态容错实践

应用属性与拓扑：Properties props = new Properties();

props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app");

props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);

props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);

props.put(StreamsConfig.producerPrefix("transaction.timeout.ms"), 60000);

StreamsBuilder builder = new StreamsBuilder();

KStream<String, String> src = builder.stream("orders");

KStream<String, Long> agg = src.groupByKey()

.count(Materialized.as("orders-count"))

.toStream();

agg.to("orders_count", Produced.with(Serdes.String(), Serdes.Long()));

KafkaStreams streams = new KafkaStreams(builder.build(), props);

streams.start();