?????? Kafka ???? ???? ????? ???? ? "InitProducerId? ???? ?? ?? ??? ???????." ??? ?????. ?? ????? ???? ?? ?? ??? ???? ???? ???? ?? ??? ?? ?? ???? ?? ??? Docker ????? ?? ???? ?? ???? ??? ???? ?? ??? ??? ??? ???? ???? ?? ??? ?????. ? ????? ? ??? ?? ??? ????? ???? ??? ???? ?? ??? ?????.

Kafka ???? ? InitProducerId ??

Kafka ????(???? API)? "??? ? ?" ??? ?? ??? ???? ?? ?????. ?? ?? ???? ??? ???? ?? ???? ????? ??? ???? ????? ???? ??? ? ????. ???? ?? ????? ???? initTransactions() ???? ???? ??? ?????. ? ???? ???? ??? ??? ID(??? ID)? ??? ??? ?? ?? Kafka ????? ???? ?????(????? ???? ?? ???? ?????)?? InitProducerId ??? ????. ? ID? ???? ??? ?????. ???? ???? ???? ???? ??? ??? ? ?? ?? InitProducerId ??? ???? ?? ?? ??? ?????.

???? ??: ???? ?? ?? ? ???? ??

?? ???? ??? ?? ?? ??? ???? ?? KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR ? KAFKA_TRANSACTION_STATE_LOG_MIN_ISR? ?? ???? ?? Kafka ?? ????? ?????. ??? ????? ??? Kafka ????? ???? ????? ?? ?????.

• KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: ???? ?? ?? ??? ?? ??? ?? ???? ??? ?? ??? ?????.
• KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: ???? ?? ?? ??? ???? ???? ?? ??, ????? ???? ?? N? ??? ???? ?????? ???.

??? ??? ??? ???? ???? ??? ?? ?? ?? ??? ?? ???? ??? ?? ? ?? ??? ?? ??? ????. InitProducerId ??? ???? ?? ?? ??? ??? ??, ? ???? ??? ?? ?????? ???? ???? ???? ???? ?? ??? ?? ???? ???? ?????? ???? ??? ????.

?? ??: ?????? Kafka? ?? ???? ??

Docker? ?? ?????? ???? Kafka? ???? ?? ???? ??? ????? ???(?? ?? ?????)? ????? ?????. Kafka? KAFKA_LISTENERS? ?? ?? ?????? ??? ????, KAFKA_ADVERTISED_LISTENERS? ?? ?? ?????(??? ??)?? ??? ?? ??? ?????.

?? Docker ????:

 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://mytestvm:9093,BROKER://localhost:9092
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092

???? ? ?? ???? ???? ????.

1. PLAINTEXT://mytestvm:9093: ?? ????? ??? ??? ? ??? -p 9093:9093? ?? ????? ???????.
2. BROKER://localhost:9092: KAFKA_INTER_BROKER_LISTENER_NAME?? ???? ????? ??? ? ?? ? ?? ???(?: ???? ?????)? ?????. KAFKA_LISTENERS? Kafka? ???? ??? 0.0.0.0:9092? ????? ????? ???? ?? ????? ?? ???? localhost:9092(ADVERTISED_LISTENERS?? ??)? ??? ???? ? ??? Docker ???? ???? ??? ???? ??? ??? ?????.

???? ????? ????? ???? KAFKA_ADVERTISED_LISTENERS? ??? ???? ???? ???? ????. ???? ?????? ?? ?? ?? BROKER ??(? ???? 9092)? Docker? ?? ???? ??? ???? ??? ???? ??? ??? ? ?? ?? ??? ?????.

?? ??: ?? ??? ??? ???? ??????.

InitProducerId ?? ?? ??? ???? ??? ???? ???? ???? ?? ?? ?? ??? ???? ? ??? ???? ????. ?? ????? Kafka?? ????? ???? ??(ADVERTISED_LISTENERS?? ?? ??? ?? ?? IP? ??? ????)? ???? ???? ?? ?????.

Docker ?? ??? ??????.

 docker run -d --name kafkacontainer -p 9093:9093 -p 9092:9092 \
-e KAFKA_BROKER_ID=1 \
-e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
-e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
-e KAFKA_ZOOKEEPER_CONNECT=172.17.0.2:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://mytestvm:9093,BROKER://localhost:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9093,BROKER://0.0.0.0:9092 \
-e KAFKA_INTER_BROKER_LISTENER_NAME=BROKER \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=???:PLAINTEXT,PLAINTEXT:PLAINTEXT \
-e KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS=1 \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
??/cp-kafka:7.0.1

? ??: -p 9092:9092? ???????. ?? ???? ??? ?? 9092? ???? ?? 9092? ?????. ?? ???(?: ??? ????? ?? ?)? localhost:9092? ??? ???? Docker? ?? ??? ?? ???? ??? Kafka Broker? ???? ??????.

??? ?? ?(???):

 org.apache.kafka.clients.producer.KafkaProducer ????;
org.apache.kafka.clients.producer.ProducerConfig ????;
java.util.Properties ????;

?? ??? TransactionalProducerExample {

    ?? ?? ?? ??(String[] args) {
        ?? props = ? ??();
        // BOOTSTRAP_SERVERS? ?? ??? Kafka Broker? ????? ?????.
        // ???? ????? ?? ??? 9093? ??? ?? 9093? ??? ? ????.
        // ???? ?????? 9092? ?? ???? ?? ???? 9092? ????? ? ? ??? ????? BOOTSTRAP_SERVERS? ?? ??? ???? ?????. // Kafka ?????? ADVERTISED_LISTENERS props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093")? ???? ?? ??? ??? ??? ?????. 
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-unique-transaction-id");
        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<string string> ??? = null;
        ???? {
            ??? = ??? KafkaProducer(props);

            // ???? ???? ???? ?? ???? ????? producer.initTransactions()? ????? ??? ? ????.
            System.out.println("Kafka ??? ???? ???? ??????!");

            //?: ???? ??, ??, ?? producer.beginTransaction();
            producer.send(new ProducerRecord("my-topic", "key1", "value1"));
            producer.send(new ProducerRecord("my-topic", "key2", "value2"));
            producer.commitTransaction();
            System.out.println("??? ????? ???????.");

        } ??(?? e) {
            System.err.println("???? ?? ??: " e.getMessage());
            if (??? != null) {
                producer.abortTransaction(); // ???? ????? ?????. System.err.println("????? ???????.");
            }
            e.printStackTrace();
        } ????? {
            if (??? != null) {
                ???.??();
            }
        }
    }
}</string>

?? ?? ? ?? ??

1. KAFKA_ADVERTISED_LISTENERS? ???: ?? ?? ??? ?????. ?????? ???? Kafka Broker? ????? ??? ???? ???? ???. ???? ??? ????? ???? localhost? ?? ????? ?? BROKER://localhost:9092? ?????. ???? ?? ????? ?? ???? ?? IP/??? ??? ?? ????? ?? localhost? ?? ??? ??? ??? ??? ???.
2. ?? ??? ????: ? ?? ?? ??? ????? ???? ???? Kafka ????? ????? ????? ???? ?? ?? ??? ????? ?????. ?? KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR, KAFKA_TRANSACTION_STATE_LOG_MIN_ISR ?? ???? ?? ? ???? ????? ?? ???? ????? ??(?? ?? 3/2? 3?? ???? ???? ????? ?? 2?? ???? ???? ????).
3. ???? ?? ???: ??? ???? ???? ??(?: Telnet ?? nc)? ???? ???? ?? ???? Kafka Broker? ??(?: Telnet localhost 9092)? ???? ???? ??? ??? ? ????.
4. Kafka ??: Kafka Broker ??(docker ?? kafkacontainer)? ?????. ???? ?? ??, ?? ?? ?? ?? ??? ??? ?? ??? ??? ??? ? ????.

??

InitProducerId ??? ???? ?? 60000??? ?? ??? ?? ??? ?????? Kafka ???? ????? ?? ???? ?????. ?? ??? ?? ???? ??? ?? ??? ??? ???? ???? ?????? ???? ??? ?? Docker ????? ???? ???? ?? ? ?? ?????. ? ??? ??? ?? Kafka ??? ??(?? ?? ??? ???? ??)? ?????? ???? ???? ???? Kafka? ???? ??? ??????? ????? ??? ? ????.

? ??? ?????? Kafka ???? ??? ?? ?? ??: InitProducerId ?? ??? ?? ?????. ??? ??? PHP ??? ????? ?? ?? ??? ?????!