Apache Kafka ‐ Java 기반 Producer 구현 및 Producer 내부 메커니즘 이해 - thought-corner/Backend-PlayGround GitHub Wiki

@Configuration
public class KafkaProducerConfig {

    private static final String TOPIC = "simple-topic";

    @Bean
    public Properties kafkaProperties() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "Hello, Kafka!");
        producer.send(record);
        producer.flush();
        producer.close();
        return properties;
    }
}

@Configuration
public class KafkaProducerConfig {

    private static final String TOPIC = "simple-topic";

    @Bean
    public Properties kafkaProperties() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return properties;
    }

    @Bean
    public CommandLineRunner commandLineRunner(Properties kafkaProperties) {
        return args -> {
            KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProperties);
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "my-key", "Hello, Kafka with key!");
            producer.send(record);
            producer.flush();
            producer.close();
        };
    }
}

@Configuration
public class KafkaProducerConfig {

    private static final String TOPIC = "simple-topic";

    @Bean
    public Properties kafkaProperties() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return properties;
    }

    @Bean
    public CommandLineRunner commandLineRunner(Properties kafkaProperties) {
        return args -> {
            KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProperties);
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "my-key", "Hello, Kafka with implemented callback!");

            // Callback 인터페이스를 직접 구현한 익명 클래스 사용
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        // 성공 처리
                        System.out.println("Message sent successfully to topic " + metadata.topic() +
                                " partition " + metadata.partition() +
                                " with offset " + metadata.offset());
                    } else {
                        // 실패 처리
                        System.err.println("Failed to send message: " + exception.getMessage());
                    }
                }
            });

            producer.flush();
            producer.close();
        };
    }
}

@Configuration
public class KafkaProducerConfig {

    private static final String TOPIC = "simple-topic";

    @Bean
    public Properties kafkaProperties() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        return properties;
    }
}

// 카프카에게 보낼 키/값 쌍. 이는 레코드가 전송되는 주제 이름, 선택적 파티션 번호, 그리고 선택적 키와 값으로 구성됩니다.
// 유효한 파티션 번호가 지정되면 레코드를 보낼 때 해당 파티션이 사용됩니다. 파티션이 지정되지 않았지만 키가 존재한다면, 키의 해시를 사용해 파티션을 선택합니다. 키와 파티션이 모두 없으면 라운드로빈 방식으로 파티션이 할당됩니다. 파티션 번호는 0-인덱스임을 유의하세요.
// 레코드에는 타임스탬프도 함께 붙어 있습니다. 사용자가 타임스탬프를 제공하지 않았다면, 제작자는 현재 시간을 기록에 스탬핑합니다. 카프카가 최종적으로 사용하는 타임스탬프는 해당 주제에 맞게 설정된 타임스탬프 유형에 따라 달라집니다.
// CreateTime설정되어 있다면, 프로듀서 레코드의 타임스탬프는 브로커가 사용합니다.
// LogAppendTime설정되어 있다면, 프로듀서 레코드의 타임스탬프는 브로커가 메시지를 로그에 추가할 때 브로커의 현지 시간으로 덮어쓰게 됩니다.
public class ProducerRecord<K, V> {

    private final String topic;
    private final Integer partition;
    private final Headers headers;
    private final K key;
    private final V value;
    private final Long timestamp;

    // ...
}

@Component
public class CustomPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (keyBytes == null) {
            // 키가 없는 경우, 라운드 로빈 방식으로 파티션 선택 (DefaultPartitioner 동작 유사)
            return Utils.toPositive(Utils.murmur2(valueBytes)) % numPartitions;
        }

        if ("special-key".equals(key)) {
            return numPartitions - 1;
        }

        // 그 외의 모든 메시지는 키의 해시값을 기반으로 파티션을 결정합니다. (DefaultPartitioner 동작)
        return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
    }

    @Override
    public void close() {
        // 리소스 정리
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 설정 초기화
    }
}