본문으로 건너뛰기

Apache Kafka Driver — 기술 레퍼런스

PlantPulse Edge 의 Apache Kafka 드라이버는 kafka-clients 4.1.2 라이브러리를 래핑한 자바 구현입니다. 외부 streaming platform (Apache Kafka / Confluent Cloud / AWS MSK / Azure Event Hubs Kafka API) 의 topic 을 consumer 로 subscribe 하여 record 를 in-memory 캐시에 보관하고, 태그의 polling cycle 마다 캐시값을 반환합니다. write 는 producer.send (async).

소스: plantpulse.driver.protocol.kafka.*

클래스책임
KafkaDriverProtocolDriver 구현체 — connect/read/write/close, topic 캐시, lazy subscribe, background poller thread

라이브러리: lib/kafka-clients-4.1.2.jar

MQTT / WebSocket 과의 차이​

PlantPulse 는 세 가지 message-based 드라이버 를 제공합니다 — 모두 동일한 JsonExtract 4-mode JSON 디코더를 공유합니다.

드라이버wire protocolpayload사용 시점
MQTT_CLIENTMQTT v3.1.1임의 (보통 JSON/평문)IIoT broker (HiveMQ / Mosquitto / EMQX), QoS 0/1/2
WEBSOCKETws/wss임의 (보통 JSON)streaming server / proprietary push API
KAFKA (이 페이지)Kafka wire protocolrecord key/value (보통 JSON String)high-throughput streaming, replay (offset), consumer group
SPARKPLUG_BMQTT + ProtobufSparkplug B PayloadCirrus Link / Tahu / Ignition 표준 (MQTT 위 IIoT spec)

Kafka 는 persistent log + offset replay 가 핵심 — 신규 consumer 가 auto-offset=earliest 로 등록하면 지난 데이터부터 받을 수 있습니다 (broker 의 retention 안에서).

동작 흐름​

broker ──ConsumerRecord──> KafkaConsumer.poll() ──> pollerLoop (background thread)
                                                       │
                                                       ├─ rawByTopic.put(topic, value)
                                                       └─ JsonExtract.updateCache(topicCache, topic, value)

PlantPulse 폴러 ──read(addr)──> JsonExtract.parse(addr.address) ──> JsonExtract.extract(...)   (in-memory hit)
                  write(addr, value) ──KafkaProducer.send(ProducerRecord(topic, value))──> broker

read 의 첫 호출 시 cache miss 면 lazy subscribe (consumer.subscribe(...) 호출) 후 빈 문자열 반환 — 다음 cycle 부터 실제 값이 들어옵니다. tag_map 에 등록된 topic 들은 connect 직후 preSubscribeFromTagMap() 으로 일괄 subscribe 되어 첫 cycle 에 바로 값이 도착할 수도 있습니다.

4-mode JSON 디코딩 spec​

JsonExtract 유틸리티가 MQTT / WebSocket / Kafka 세 driver 의 메시지 디코딩을 통일합니다.

모드address 형식동작
SCALAR<topic> 또는 <topic>.value메시지 전체를 String 으로. JSON object 면 raw 폴백.
KEY<topic>:<json-key>top-level JSON object 의 한 key 값
PATH<topic>:$.<json-path>JSON Pointer 동적 evaluate (예: $.data.tags.T1 → /data/tags/T1)
RAW<topic>:_raw_마지막 raw 메시지 (디버깅)

cache 구조 — 단일 Map<String, String> 사용, key 는 topic + "|" + field:

key의미
topic|_raw_마지막 raw 메시지
topic|valuescalar (JSON 아닌 메시지일 때만)
topic|<json-key>top-level JSON key 별 값 (object 일 때만)

PATH 모드는 cache 를 거치지 않고 rawByTopic 의 마지막 raw 메시지를 매번 JSONPointer.queryFrom(...) 로 evaluate.

옵션​

키기본값설명
group-idplantpulse-edge-<opc_id>consumer group id. offset commit 단위.
security-protocolPLAINTEXTPLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL
sasl-mechanism(없음)PLAIN / SCRAM-SHA-256 / SCRAM-SHA-512
sasl-jaas-config(없음)JAAS 설정 (예: ...PlainLoginModule required username="u" password="p";)
auto-offsetlatestearliest / latest — 신규 group 의 시작점

enable.auto.commit=true 가 항상 설정되어 broker 가 자동으로 offset 을 commit 합니다 (기본 5초 주기).

연결 라이프사이클​

메서드동작
connect()options 적용 → consumer / producer 생성 → poller thread 시작 → tag_map 일괄 subscribe
pollerLoop()consumer.poll(500ms) 반복 → record 별 JsonExtract.updateCache(...)
read()parse → cache miss 면 lazy subscribe → cache hit 면 값 반환
write()producer.send(new ProducerRecord<>(topic, value)) (async)
close()consumer.wakeup() → thread join → consumer/producer close → cache clear

한계 및 향후 작업​

  • 재연결 — kafka-clients 는 자체 재연결 로직을 갖지만, broker 전체가 down 되어 long-lived poll() 이 실패할 때 driver 측에서 setConnected(false) 를 별도로 마킹하지 않음 → 운영 상태 모니터링은 broker side metric 권장.
  • AWS MSK IAM — aws-msk-iam-auth jar 추가 필요 (현재 미포함). 추후 aws-msk-iam-auth-1.x.jar 를 lib/ 에 넣고 sasl.client.callback.handler.class 를 옵션으로 노출 검토.
  • TLS truststore — 드라이버 옵션으로는 지정할 수 없고 JVM 시스템 truststore 를 씁니다. 자체서명 사설 CA 를 쓰는 broker 라면 CA 인증서를 JVM 의 truststore 에 넣거나, 게이트웨이 기동 옵션에 -Djavax.net.ssl.trustStore=… 를 추가하세요.
  • partition / key 지정 producer — write 는 항상 key 없이 보내므로 파티션이 round-robin 으로 정해집니다. 즉 같은 태그의 쓰기가 순서대로 도착한다는 보장이 없습니다. 파티션 단위 순서가 필요하면 토픽을 파티션 1개로 만들거나, 순서 보장을 consumer 쪽에서 처리하세요.
  • transactional producer — 미지원. 현재 idempotent producer 도 명시 enable 안 함.

테스트 위치​

test/java/plantpulse/driver/protocol/kafka/KafkaDriverTest.java — 18 테스트.

  • 클래스 로드 / 계층 (BaseProtocolDriver / ProtocolDriver)
  • 초기 상태 / 메타 (isConnected/isWriteSupported/isExternal/getDriverSource)
  • 미연결 read/write/null 안전
  • getDebugBaseUrl — kafka://, default 포트 (9092)
  • 옵션 파싱 — group-id / security-protocol / sasl-* / auto-offset, options=null 안전
  • 4-mode 디코딩 — cache 직접 채워서 KEY / PATH / RAW / SCALAR / .value suffix 모두 검증
  • close() 미연결 안전

test/java/plantpulse/driver/protocol/common/JsonExtractTest.java — 23 테스트 (4-mode parser/cache/extract 통합 검증).

실 broker 의존 통합 테스트는 별도 — 본 단위 테스트는 외부 Kafka 없이 동작합니다.