[Kafka] Spring-Kafka API 적용 & 테스트

Kafka version : 3.0.10

 

spring-kafka 공식문서를 많이 읽어보자.

https://docs.spring.io/spring-kafka/reference/index.html

Overview :: Spring Kafka

 

- 전송 ( Producer )

 

메세지를 전송하는건 KafkaTemplate 클래스를 이용할 수 있다.

0. 설정

@Bean
public ProducerFactory<Integer, String> producerFactory() {
    return new DefaultKafkaProducerFactory<>(producerConfigs());
}

@Bean
public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    // See https://kafka.apache.org/documentation/#producerconfigs for more properties
    return props;
}

@Bean
public KafkaTemplate<Integer, String> kafkaTemplate() {
    return new KafkaTemplate<Integer, String>(producerFactory());
}

KafkaTemplate을 사용하기 위해선, producerFactory를 빈으로 등록하고 생성자로 제공한다.

 

 

 

1. 전송

 

kafkaTemplate을 선언해주자.

private final KafkaTemplate<String, String> kafkaTemplate;

 

Kafka의 send메서드를 살펴보겠다.

 

- KafkaTemplate.java

CompletableFuture<SendResult<K, V>> sendDefault(V data);

CompletableFuture<SendResult<K, V>> sendDefault(K key, V data);

CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, K key, V data);

CompletableFuture<SendResult<K, V>> sendDefault(Integer partition, Long timestamp, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, V data);

CompletableFuture<SendResult<K, V>> send(String topic, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, V data);

CompletableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);

CompletableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record);

CompletableFuture<SendResult<K, V>> send(Message<?> message);

 

topic과 함께 보내는 메서드를 보면, 첫 번째 파라미터로는 topic을 String형식으로 전달하는건 고정이다.

 

두 번째 파라미터부터 partition, timestamp, key, data를 전달할 수 있다.

 

timestamp를 같이 전달하면, 레코드 내에 시간을 같이 담게 된다.

 

해당 topic의 설정에 따라 다르다.

• CREATE_TIME 으로 설정된 경우 : 사용자가 전달한 timestamp로 기록되거나, 전달되지 않으면 자동으로 생성된다.
• LOG_APPEND_TIME 으로 설정된 경우 : 사용자가 전달한 timestamp는 무시된다. 브로커가 로컬 시간을 넣게된다.

 

 

 

 

이전 포스팅에서 다음과 같은 내용이 있었다.

 

프로듀서가 만약 Topic 0에 메세지를 넣으려고 시도할 때, 라운드 로빈Round Robin 알고리즘으로 파티션을 선택하거나, 파티션의 키Key(이는 프로듀서가 메세지를 저장할 때 토픽과 같이 전달해야 한다.)를 통해 저장할 Partition을 선택하게 된다.

 

여기서 프로듀서가 전달하는 키를 전달하며 저장할 Partition을 지정하게 된다.

 

키를 전달하면 키를 해싱하여 결과에 따라 파티션에 저장하게 된다.

 

키를 사용하려면 KafkaConfig에 다음 설정을 추가해야한다.

...

    @Value("${kafka.topic-key}")
    public String TOPIC_WITH_KEY;

    @Autowired
    private KafkaAdmin kafkaAdmin;

    private NewTopic defaultTopic() {
        return TopicBuilder.name(DEFAULT_TOPIC)
                .partitions(2)
                .replicas(2)
                .build();
    }

    private NewTopic topicWithKey() {
        return TopicBuilder.name(TOPIC_WITH_KEY)
                .partitions(2)
                .replicas(2)
                .build();
    }

    @PostConstruct
    public void init() {
        kafkaAdmin.createOrModifyTopics(defaultTopic());
        kafkaAdmin.createOrModifyTopics(topicWithKey());
	}

...

Key는 다음에 적용해 보도록 하고, 전송하는 예제 코드를 보자.

 

1-1. 비동기 전송 (Non-Blocking)

public void sendToKafka(final MyOutputData data) {
    final ProducerRecord<String, String> record = createRecord(data);

    CompletableFuture<SendResult<Integer, String>> future = template.send(record);
    future.whenComplete((result, ex) -> {
        if (ex == null) {
            handleSuccess(data);
        }
        else {
            handleFailure(data, record, ex);
        }
    });
}

Java의 CompletableFuture을 통해 비동기 처리에 대한 결과값을 받을 수 있다.

(현재 버전에서는 CompletableFuture를 반환하고 있지만, 다른 버전에서는 ListenableFuture로 반환하는 것 같다.)

이후에 Suceess, Failure를 handle하는 메서드는 생략하겠다.

 

1-2. 동기 전송(Blocking)

public void sendToKafka(final MyOutputData data) {
    final ProducerRecord<String, String> record = createRecord(data);

    try {
        template.send(record).get(10, TimeUnit.SECONDS);
        handleSuccess(data);
    }
    catch (ExecutionException e) {
        handleFailure(data, record, e.getCause());
    }
    catch (TimeoutException | InterruptedException e) {
        handleFailure(data, record, e);
    }
}

동기 전송도 마찬가지로 CompletableFuture를 반환하기 때문에 get메서드를 통해 handle할 수 있다.

 

 

 

1-3. ProducerRecord클래스

 

ProducerRecord클래스는 Kafka에서 제공하는 클래스이다. 이를 이용하여 send메서드의 파라미터들을 담을 수 있다.

아래는 ProducerRecord클래스의 필드값들이다.

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;
    
    ...
    
   }

send에 필요한 값들을 필드 값으로 갖고있다.

 

여러 블로글를 보았을 때 간단하게 topic을 정해놓고 message만 전달하고 있었다.

위와 같이 ProducerRecord를 이용해서 매개변수들을 정형화하고 controller나 다른 서비스에서 처리하는게 좋아보인다.

 

 

 

- 수신 ( Consumer )

 

0. 설정

 

@Configuration
@EnableKafka
public class KafkaConfig {

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
                        kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
        ...
        return props;
    }
}

@KafkaListner를 사용하기 위해, @EnableKafka를 활성화시켜주어야 한다.

 

1. 수신

메세지를 수신하기 위해선 @KafkaListner 어노테이션을 사용한다.

 

1-1. 기본 수신

public class Listener {

    @KafkaListener(id = "foo", topics = "myTopic", clientIdPrefix = "myClientId")
    public void listen(String data) {
        ...
    }

}

필요한 매개변수들을 설정하여 data를 listen할 수 있다.

 

여기서 Kafka의 특징인 Event Driven의 성격을 잘 볼 수 있다.

Consumer는 메세지가 왔는지 DB를 접근하는 등 수시로 확인하는게 아니라, 메세지를 받은 Event가 발생함에 따라 실행된다.

 

1-2. 토픽, 파티션 지정 수신

@KafkaListener(id = "thing2", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
          @TopicPartition(topic = "topic2", partitions = "0",
             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}

위와 같이 특정 토픽이나 파티션을 지정하여 사용할 수 있다.

 

Producer가 어느 토픽과 파티션에 넣겠다고 약속해 두고 사용하면 되겠다.

 

@KafkaListener(id = "thing3", topicPartitions =
        { @TopicPartition(topic = "topic1", partitions = { "0", "1" },
             partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}

위와 같이 parition = "*" 처럼 wildcard로 사용할 수 있다.

 

@KafkaListener(id = "thing3", topicPartitions =
        { @TopicPartition(topic = "topic1",
             partitionOffsets = @PartitionOffset(partition = "0-5", initialOffset = "0"))
        })
public void listen(ConsumerRecord<?, ?> record) {
    ...
}

위와 같이 파티션의 범위를 지정하여 사용할 수도 있다.

 

1-3. 레코드의 metadata수집

@KafkaListener(id = "qux", topicPattern = "myTopic1")
public void listen(@Payload String foo,
        @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) Integer key,
        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
        @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
        ) {
    ...
}

메세지로부터 받은 키, 파티션, 토픽 등등을 받을 수 있었다.

 

 

하지만 2.5 버전부터는, 

@KafkaListener(...)
public void listen(String str, ConsumerRecordMetadata meta) {
    ...
}

 

ConsumerRecordMetadata 클래스를 매개변수로 한꺼번에 받아볼 수 있다.

 

- ConsumerRecordMetadata.java

public class ConsumerRecordMetadata {

	private final RecordMetadata delegate;

	private final TimestampType timestampType;

	...

	public boolean hasOffset() {...}

	public long offset() {...}

	public boolean hasTimestamp() {...}

	public long timestamp() {...}

	public int serializedKeySize() {...}

	public int serializedValueSize() {...}

	public String topic() {...}

	public int partition() {...}

	public TimestampType timestampType() {...}

	@Override
	public int hashCode() {...}

	@Override
	public boolean equals(Object obj) {...}

	@Override
	public String toString() {...}

}

필요한 데이터를 클래스에서 가져올 수 있겠다.

 

 

 

- 적용

 

이제 Spring 프로젝트 예제에 Kafka를 적용시켜 테스트해 보겠다.

 

0. 설정

dependency에 다음을 추가해준다.

implementation 'org.springframework.kafka:spring-kafka'

 

application.yml에 다음을 추가해준다.

spring:
  kafka:
    bootstrap-servers: "localhost:9092"
    consumer:
      group-id: "myGroup"

 

DB없는 테스트 애플리케이션이다.

@SpringBootApplication(exclude={DataSourceAutoConfiguration.class})
public class KafkaApplication {

	public static void main(String[] args) {
		SpringApplication.run(KafkaApplication.class, args);
	}

}

DataSrouce를 제외하고 실행하자.

 

1. KafkaConfig

 

- KafkaConfig.java

@EnableKafka
@Configuration
public class KafkaConfig {

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Value(value = "${spring.kafka.consumer.group-id}")
    private String groupId;



    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        return new KafkaAdmin(configs);
    }


    // -------------------------------------- Producer --------------------------------------

    // New Topic
    @Bean
    public NewTopic myTopic1() {
        return new NewTopic("topic_1", 1, (short) 1);
    }

    // New Topic 2
    @Bean
    public NewTopic myTopic2() {
        return new NewTopic("topic_2", 1, (short) 1);
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    // -------------------------------------- Consumer --------------------------------------

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

}

테스트용 토픽 topic_1, topic_2를 만들었다. yml에 설정해둔 kafka address와 매핑시켜주고, Producer와 Consumer 설정을 해두었다.

 

 

2. KafkaController

 

- KafkaController.java

@RestController
@RequestMapping(value = "/kafka")
public class KafkaController {

    @Autowired
    private KafkaService kafkaService;

    @PostMapping("")
    public String sendMessage(@RequestParam("message") String message) {
        kafkaService.sendMessage(message);
        return "Send Success!";
    }
}

url은 단순하게 /kafka로 설정했고, 사용자가 메세지를 전송하려고 시도하는 API를 작성했다.

 

매개변수 message로 전달하려는 메세지를 전달하면 된다.

 

3. KafkaService

 

 - KafkaService.java

@Slf4j
@Service
public class KafkaService {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;


    //----------- Producer -----------
    public void sendMessage(String message) {
        log.info("[Producer] Message Send : " + message);
        this.kafkaTemplate.send("topic_1", message);
    }


    //----------- Consumer -----------
    @KafkaListener(topics = "topic_1", groupId = "myGroup")
    public void consumeMyopic1(@Payload String message, ConsumerRecordMetadata metadata) {
        log.info("[Consumer] Message Received : " + message);
        log.info("[Consumer] Partition : " + metadata.partition());
    }
}

Producer가 메세지를 send하기 위해 KafkaTemplate을 선언했다.

 

Producer의 경우 message를 받아서 단순히 send하고, 설정해두었던 토픽(topic_1 )으로 메세지를 보낸다.

 

Consumer의 경우 KafkaListener 어노테이션을 사용해서 message를 listen한다.

메세지를 받으면 log를 출력하도록 해두었다.

 

 

 

 

- 테스트

 

이제 테스트를 진행해보자.

메세지를 보냈고, Reponse 200과 함께 Send success! 문구를 받을 수 있었다.

 

log도 잘 출력된 것을 확인했다. 메세지 전송에 성공했다. 

 

 

 

 

적용 코드는 깃허브에 업로드해 두었다.

https://github.com/DDonghyeo/kafka-spring

 

 

Kafka를 공부하고 Spring에 직접 적용시켜서 사용해보았다.

 

많은 내용이 있었지만 현재는 단순 메세지 전송만 테스트 해보았다.

 

다음 포스팅부턴 공식 문서에 있는 여러 레퍼런스를 활용해서 토픽&파티션 관리, 키 사용, Producer, Consumer 설정 등을 직접 적용해보고, 더욱 실용적으로 코드를 정형화해 보겠다.

 

참고
https://docs.spring.io/spring-kafka/reference/
https://spring.io/projects/spring-kafka
https://happy-jjang-a.tistory.com/201
https://devocean.sk.com/blog/techBoardDetail.do?ID=164096