KStreamJoinKTable 에서
344
작성한 질문수 4
ADDRESS_TABLE에서 기존의 key에 주소를 새로 작성했을 때에는, consumer에 메시지가 새로 뜨지 않고, ORDER_STREAM에서 기존의 key에 새로운 value를 작성했을 때에만 consumer에 메시지가 새로 뜨는 이유는 무엇인가요?
답변 2
0
안녕하세요. ADDRESS_TABLE은 KTable로 선언하고 ORDER_STREAM은 KStream으로 선언하여 활용하는 예제를 말씀하시는 것으로 이해되었는데요. 기본적으로 KStream, KTable join에서는 KStream의 데이터 인입을 기준으로 처리를 수행하기 때문에 ADDRESS_TABLE에 해당하는 토픽에 레코드를 전송하더라도 따로 출력되는 내용은 없다고 보면 됩니다.
0
안녕하세요, 인프런 AI 인턴이에요.
KStreamJoinKTable은 Kafka Streams의 API 중에 하나로, KStream과 KTable을 조인하는 작업을 수행할 수 있습니다.
제가 이해한 내용으로는, 주소를 새로 작성했을 때에는 ADDRESS_TABLE의 key에 변화가 생겨서 조인 결과가 변경되지 않았기 때문에 consumer에 메시지가 새로 뜨지 않는 것 같아요.
반면, ORDER_STREAM에서 기존의 key에 새로운 value를 작성했을 때는 조인 결과가 변경되어 consumer에 메시지가 새로 뜨는 것 같습니다. 이 경우에는 조인 결과가 변경됨에 따라 새로운 데이터가 생성된 것이기 때문에 consumer가 이를 감지하고 메시지를 받는 것으로 이해하고 있습니다.
더 정확한 이유를 알기 위해서는 코드나 설정 상세 내용 등을 더 자세히 알아야 할 것 같습니다. 좀 더 구체적인 정보를 제공해주실 수 있다면 더 도움을 드릴 수 있을 것 같아요.
consume 이후 lag가 줄어들지 않음
0
69
2
안녕하세요. 강의의 카프카 버전과 현재 시점의 카프카의 차이점 문의 드립니다.
0
102
2
멱등성 프로듀서 retries 관련 질문입니다.
0
95
2
채팅 서비스 개발 시 주의점이 있을까요?
0
95
2
충분히 큰 파티션 생성시 궁금증이 존재합니다.
0
83
2
KTable 키가 없는 레코드 처리
0
70
2
컨슈머 테스트 코드 작성
0
92
2
리밸런스 onPartitionRevoked이 필요한 상황
0
72
2
카프카 클러스터에서 감당 가능한 파티션(레플리카) 수 문의
0
117
2
reset offset 질문
0
73
2
KStreamJoinKTable 실행시 오류
0
74
2
auto.commit.interval.ms 옵션 관련 질문 드립니다.
0
137
2
파티션, 컨슈머 그룹, 컨슈머 관련 질문
0
116
1
java, kotlin
0
126
2
shutdownThread 에 대한 문의 입니다.
0
150
2
zookeeper실행시 오류가 발생합니다.
0
265
2
커스텀 소스 커넥터에서 Thread.sleep (1000) 은 왜 하는거에요?
0
149
2
처리량을 늘리기 위해서 파티션을 늘리고 컨슈머를 늘려야한다고 설명하셨는데요
0
178
3
파티션 개수와 컨슈머 개수의 처리량 관련 질문
0
162
2
동영상 및 이미지 처리 관련 문의 드립니다.
0
221
2
주키퍼 없이 사용 문의 드립니다.
0
296
1
kafka 를 띄우니 오류가 발생하고 종료합니다.
0
278
3
zookeeper 실행시 오류 .. 무엇을 해야 할까요?
0
216
2
파티션 추가로 해결할 수 있지만 늘어난 파티션은 줄일 수 없지 않나요?
0
192
2