[kafka] 데이터 producer & consumer test

[구성 참고 사이트 및 기타 출처] https://abbf.tistory.com/25 의 표 내용 참고

[구성 및 설정] https://abbf.tistory.com/26

[topic 생성 및 확인] https://abbf.tistory.com/27 

 

 

기본 개념에 대한 내용은 적지 않고 테스트한 내용만 기록.

 

데이터 producer & consumer 를 수행하려면 아래 두가지를 사용한다. (위치 : /kafka/bin)

 - /kafka-console-producer.sh

 - /kafka-console-consumer.sh

 

1) producer

수행 서버와 어느 토픽에 넣을지 옵션을 주고 엔터를 치면 입력받는 형태로 나온다.

scala 2.12 테스트 할 때 bootstrap-server를 한쪽만 주고 해서 파티션에 데이터가 하나만 남았다.

실제 생각하기에 이게 replication이라 한쪽에서만 쏴도 당연히 데이터가 복제가 되는줄 알았던거라서..

(ex. ./kafka-console-producer.sh --bootstrap-server master.com:9092 --topic NewTopic2)

그래서 이번에 할때는 bootstrap-server에 서버 2개 주소를 둘 다 주고 테스트를 했다.

./kafka-console-producer.sh --bootstrap-server master.com:9092,slave1.com:9092 --topic SetTopic3

그랬더니 똑같이 데이터가 남음.

이게 정상임....

server1 (master.com)
server2 (slave1.com)

 

 

**비고

파티션 들어가있는 디렉토리에 bk 등 백업디렉토리 만들어 두면 kafka가 에러 뱉고 죽는다.

[2024-01-12 15:18:55,473] ERROR Exiting Kafka due to fatal exception during startup. (kafka.Kafka$) org.apache.kafka.common.KafkaException: Found directory /kafka/kaf_log/bk, 'bk' is not in the form of topic-partition or topic-partition.uniqueId-delete (if marked for deletion). Kafka's log directories (and children) should only contain Kafka topic data.         at kafka.log.LocalLog$.exception$1(LocalLog.scala:685)         at kafka.log.LocalLog$.parseTopicPartitionName(LocalLog.scala:690)         at kafka.log.LogManager.$anonfun$loadLogs$10(LogManager.scala:417)         at kafka.log.LogManager.$anonfun$loadLogs$10$adapted(LogManager.scala:412)         at scala.collection.ArrayOps$.filter$extension(ArrayOps.scala:564)         at kafka.log.LogManager.$anonfun$loadLogs$4(LogManager.scala:412)         at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:576)         at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:574)         at scala.collection.AbstractIterable.foreach(Iterable.scala:933)         at kafka.log.LogManager.loadLogs(LogManager.scala:377)         at kafka.log.LogManager.startupWithConfigOverrides(LogManager.scala:567)         at kafka.log.LogManager.startup(LogManager.scala:527)         at kafka.server.KafkaServer.startup(KafkaServer.scala:282)         at kafka.Kafka$.main(Kafka.scala:113)         at kafka.Kafka.main(Kafka.scala) [2024-01-12 15:18:55,474] INFO [KafkaServer id=1] shutting down (kafka.server.KafkaServer)