해당 포트 제한되었는지 확인하여 풀어줘야함 (ex 4369 port)-확인: netsh interface ipv4 show excludedportrange protocol=tcp-해제:net stop winnatnetsh int ipv4 add excludedportrange protocol=tcp startport=4369 numberofports=1net start winnat

bin 폴더에서grafana-server start

이미 Kafka Core편은 들었고 현재 Kafka Connect부분중 Sink Connector듣기 전입니다.이 강의를 듣게된 동기이기도 한데요.. DBToDB방식으로 해결하지 않고 Source Connector가 없이 Topic를 만들고 Topic에 Produce하는 방법이 가능한지요? Topic에 생성된 메시지는 Sink Connector 로 DB에 적재되는 구조입니다. Topic의 메시지 구조는 Source Connector에서 생성된 Message 체계로 Topic에 메시지를 적재하는 식으로 처리할려고 합니다.Kafka core에서 배웠던 Produce하는 방법을 이용하는경우입니다. 정리) Source Connector없이 Topic에 메시지를 보내고 이를 SinkConnect로 DB에 적재하는게 가능한지요?감사합니다.

kafka가 뭔지 잘 몰랐는데 덕분에 잘 수강하고 있습니다.order-service로 하나의 db에 데이터동기화는 이해했습니다.제목의 질문인 멀티db의 경우는 어떻게 처리하는지 궁금해서 질문드립니다. order-service1 --- order-db1order-service2 --- order-db2  이렇게 연결된 서비스의 경우kafka로 db1에 입력된 정보가 db2로db2는 db1으로 서로 동기화도 가능한가요? 서비스가 더 증가할 경우도 가능한가요? (db1 -> db2,3,4...) db를 master/slave 구성은 해봤는데 msa로 동일db 여러개가 동기화 설정 가능한지 궁금해서 문의드립니다. 

mysql에 debezium source connector로 topic에 가져온 데이터를 받는 stream을 만들고그 stream을 기반으로 하여CREATE TABLE timeout WITH (KAFKA_TOPIC='timeout' , KEY_FORMAT='AVRO', VALUE_FORMAT='AVRO', PARTITIONS=1) AS > SELECT > order_id -> order_id AS order_id, > TIMESTAMPADD(MILLISECONDS, 9 * 3600 * 1000, PARSE_TIMESTAMP(LATEST_BY_OFFSET(order_datetime), 'yyyy-MM-dd''T''HH:mm:ssX')) AS last_log_time > FROM orders > GROUP BY order_id -> order_id > HAVING ((UNIX_TIMESTAMP(CONVERT_TZ(FROM_UNIXTIME(UNIX_TIMESTAMP()), 'UTC', 'Asia/Seoul')) - UNIX_TIMESTAMP(TIMESTAMPADD(MILLISECONDS, 9 * 3600 * 1000, PARSE_TIMESTAMP(LATEST_BY_OFFSET(order_datetime), 'yyyy-MM-dd''T''HH:mm:ssX')))) / 1000 > 600) > EMIT CHANGES;이런식으로 id별로 마지막 로그 시간이 오고 10분 이상이 지나면 table에 담기도록 만들었습니다처음에 이미 10분이 지난 데이터를 넣으면 table에도 들어가고 topic에도 잘 들어가는데현재시간의 데이터를 넣고 10분이 지나면 table에는 들어가는데 topic에는 들어가지 않습니다table에도 담기고 topic에도 담기려면 어떻게 해야하나요? 아니면 원래 불가능한건가요?기반한 stream은 데이터를 넣으면 곧 바로 stream과 토픽에 잘 들어갑니다.|ORDER_ID |CALCULATED_TIME |LAST_LOG_TIME | +------------------------------------------+------------------------------------------+------------------------------------------+ |1 |78088 |2024-06-16T12:30:00.000 | |2 |69988 |2024-06-16T14:45:00.000 | |3 |72088 |2024-06-16T14:10:00.000 | |4 |32739088 |2023-06-04T12:00:00.000 | |5 |32637088 |2023-06-05T16:20:00.000 | |6 |32567788 |2023-06-06T11:35:00.000 | |7 |69058 |2024-06-16T15:00:30.000 | |8 |68698 |2024-06-16T15:06:30.000 | |9 |66958 |2024-06-16T15:35:30.000 | |10 |65698 |2024-06-16T15:56:30.000 | |11 |66298 |2024-06-16T15:46:30.000 | |12 |4258 |2024-06-17T09:00:30.000 | |13 |3418 |2024-06-17T09:14:30.000 | |14 |1918 |2024-06-17T09:39:30.000 | |15 |2429 |2024-06-17T09:30:59.000 | Query terminated ksql> print result7777; Key format: AVRO or KAFKA_STRING Value format: AVRO rowtime: 2024/06/16 04:23:23.878 Z, key: 1, value: {"CALCULATED_TIME": 12183, "LAST_LOG_TIME": 1718541000000}, partition: 0 rowtime: 2024/06/16 04:23:23.879 Z, key: 2, value: {"CALCULATED_TIME": 4083, "LAST_LOG_TIME": 1718549100000}, partition: 0 rowtime: 2024/06/16 05:10:08.498 Z, key: 3, value: {"CALCULATED_TIME": 6183, "LAST_LOG_TIME": 1718547000000}, partition: 0 rowtime: 2024/06/16 06:06:52.365 Z, key: 4, value: {"CALCULATED_TIME": 32673183, "LAST_LOG_TIME": 1685880000000}, partition: 0 rowtime: 2024/06/16 06:06:52.373 Z, key: 5, value: {"CALCULATED_TIME": 32571183, "LAST_LOG_TIME": 1685982000000}, partition: 0 rowtime: 2024/06/16 06:06:52.377 Z, key: 6, value: {"CALCULATED_TIME": 32501883, "LAST_LOG_TIME": 1686051300000}, partition: 0 rowtime: 2024/06/16 06:09:36.530 Z, key: 7, value: {"CALCULATED_TIME": 3153, "LAST_LOG_TIME": 1718550030000}, partition: 0 rowtime: 2024/06/16 06:15:08.351 Z, key: 8, value: {"CALCULATED_TIME": 2793, "LAST_LOG_TIME": 1718550390000}, partition: 0 rowtime: 2024/06/16 06:41:28.920 Z, key: 9, value: {"CALCULATED_TIME": 1053, "LAST_LOG_TIME": 1718552130000}, partition: 0 rowtime: 2024/06/17 00:23:09.442 Z, key: 12, value: {"CALCULATED_TIME": 1372, "LAST_LOG_TIME": 1718614830000}, partition: 01-9, 12 이미 10분이 지난 데이터 // 그 외 = 데이터가 mysql에 담기고 10분이 지나 table에 담긴 데이터

- 카프카를 공부하시면서 생긴 질문들을 남겨주세요. 상세히 작성하면 더 좋아요! - 먼저 유사한 질문이 있었는지 검색해보세요. - 서로 예의를 지키며 존중하는 문화를 만들어가요. - 잠깐! 인프런 서비스 운영 관련 문의는 1:1 문의하기를 이용해주세요. 안녕하세요~auto_commit_interval_ms_config 에 대해 궁금한 점이 있는데요 찾아봐도 모호해서 질문드립니다.auto_commit_interval_ms_config = 60이라고 가정했을때위 옵션은 poll() 호출 여부와 관계없이 60초 마다 자동으로 커밋을 해주는건가요?아니면 마지막 자동커밋 발생하고 60초 이후에 poll() 이 호출될때 커밋을 해준다는건가요?

강의를 따라 docker-compose 를 이용 해 kafka와 zookeeper 컨테이너를 설정을 수정하고,도커에서 docker-compose-single-broker.yml 파일을 실행하는 과정에서 에러가 발생했습니다. 최근에 올라온 유사한 질문의 답변을 참고 해 똑같이 yaml 파일을 구성 했는데, 해결 되지 않았습니다.참고: https://www.inflearn.com/questions/1187585/docker-compose-f-docker-compose-single-broker-yml-up-d https://github.com/wurstmeister/kafka-docker의 master branch clonewindows OSdocker-compose-single-broker.ymlversion: '2' services: zookeeper: image: wurstmeister/zookeeper ports: - "2181:2181" networks: my-network: ipv4_address: 172.18.0.100 kafka: # build: . image: wurstmeister/kafka ports: - "9092:9092" environment: KAFKA_ADVERTISED_HOST_NAME: 172.18.0.101 KAFKA_CREATE_TOPICS: "Topic1:1:3" KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 volumes: - /var/run/docker.sock:/var/run/docker.sock depends_on: - zookeeper networks: my-network: ipv4_address: 172.18.0.101 networks: my-network: external: true name: ecommerce-network명령어$ docker-compose -f docker-compose-single-broker.yml up -dError[+] Building 36.9s (8/9) docker:default => [kafka internal] load build definition from Dockerfile 0.0s => => transferring dockerfile: 2.40kB 0.0s => [kafka internal] load metadata for docker.io/library/openjdk:11-jre-slim 2.0s => [kafka auth] library/openjdk:pull token for registry-1.docker.io 0.0s => [kafka internal] load .dockerignore 0.0s => => transferring context: 2B 0.0s => [kafka 1/4] FROM docker.io/library/openjdk:11-jre-slim@sha256:93af7df2308c5141a751c4830e6b6c5717db102b3b31f012ea29d842dc4f 0.0s => [kafka internal] load build context 0.0s => => transferring context: 243B 0.0s => CACHED [kafka 2/4] COPY download-kafka.sh start-kafka.sh broker-list.sh create-topics.sh versions.sh /tmp2/ 0.0s => ERROR [kafka 3/4] RUN set -eux ; apt-get update ; apt-get upgrade -y ; apt-get install -y --no-install-recomm 34.8s ------ ...생략 failed to solve: process "/bin/sh -c set -eux ; apt-get update ; apt-get upgrade -y ; apt-get install -y --no-install-recommends jq net-tools curl wget ; apt-get install -y --no-install-recommends gnupg lsb-release ; \tcurl -fsSL https://download.docker.com/linux/debian/gpg | gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg ; \techo \t\t\"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/debian \t\t$(lsb_release -cs) stable\" | tee /etc/apt/sources.list.d/docker.list > /dev/null ; apt-get update ; apt-get install -y --no-install-recommends docker-ce-cli ; apt remove -y gnupg lsb-release ; apt clean ; apt autoremove -y ; apt -f install ; apt-get install -y --no-install-recommends netcat ; chmod a+x /tmp2/*.sh ; mv /tmp2/start-kafka.sh /tmp2/broker-list.sh /tmp2/create-topics.sh /tmp2/versions.sh /usr/bin ; sync ; /tmp2/download-kafka.sh ; tar xfz /tmp2/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz -C /opt ; rm /tmp2/kafka_${SCALA_VERSION}-${KAFKA_VERSION}.tgz ; ln -s /opt/kafka_${SCALA_VERSION}-${KAFKA_VERSION} ${KAFKA_HOME} ; rm -rf /tmp2 ; rm -rf /var/lib/apt/lists/*" did not complete successfully: exit code: 2 강의 내용과는 다르게 뭘 다운 받더니 중간에 ERROR 가 발생합니다.구글링을 해도 안나오고, wurstmeister/kafka-docker 이슈에도 적혀져 있는게 없는데 혹시 해결 하실 분 계실까요? 😭😭😭😭

안녕하세요 강사님 아키텍처에 궁금한 점이 생겨 질문드립니다.카프카를 사용한 이유가 채팅 서버의 상태를 비동기로 전송하기 위함이라 이해했습니다.채팅 서버가 실행/종료될 때카프카에 이벤트 발행컨트롤 서버가 이벤트 컨슈밍하며 채팅서버 상태 확인제가 제대로 이해한게 맞다면 대략 위와 같은 흐름이라 이해했습니다. 즉 채팅 서버의 상태를 컨트롤 서버에 알리기 위해 카프카라는 메시지 브로커를 둔것 같은데, 카프카를 사용하지 않고 채팅서버에서 컨트롤 서버로 직접 상태를 전송하는 방식은 불가능할까요.? 이 상황에서 카프카가 굳이 필요한건지 라는 의문이 들어 질문하게 됐습니다ㅇㅅㅇ.

db에 등록도 잘 되었는지 확인해 보기 위해서coupon 테이블 조회 해보았는데 [coupon_id], [user_id]테스트 로직에서 userId 값은 단순히 for을 통해 0부터 순차적으로 받은 뒤쿠폰 개수에 맞춰 userId 도 0~99 사이의 값이 들어 올 것이라 생각했는데예상과 다르게 103 이라는 id가 들어왔습니다이런 이유가 궁금합니다

[이 강의에서 다룰 zero-downtime-deploy를 위한 아키텍처 구조도를 알려드릴게요] 강의를 들으면서 컨트롤 서버의 필요성에 대해 설명해주셨습니다. 채팅 서버 1,2,3 이 다운될 때를 고려해서 로드 밸런싱 용으로 컨트롤 서버를 따로 둔다고 이해했는데, 컨트롤 서버가 배포되거나 다운되는 경우는 없을까요? 컨트롤 서버도 이중화로 구성을 해야하는 것인지, 다른 방법은 무엇이 있는지 궁금합니당 

제가 EXCEL파일에 있는 데이터를 카프카를 통해 db에 저장을 해야하는데 이 강의를 들으면 할 수 있을까요?

제가 구상하고있는 구조가 mysql에서 debezium source connector가 topic에 넘기고 ksqldb의 streams나 table로 재구성하여 다른 topic으로 넘긴 후mysql sink database에서 받는다. 라는걸 구상중인데요 ksqldb에서 직접 insert를 하면 json 형식이 아니라서 sink connector가 읽지 못하는거 같습니다.-- debezium.json --{ "name": "debezium", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "localhost", "database.port": "3306", "database.user": "root", "database.password": "1234", "database.allowPublicKeyRetrieval": "true", "database.server.id": "10777", "database.server.name": "debe01", "database.include.list": "debe", "table.include.list": "debe.user", "database.history.kafka.bootstrap.servers": "localhost:9092", "database.history.kafka.topic": "schema-changesde.mysql.oc", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable": "false", "value.converter.schemas.enable": "false", "database.connectionTimeZone": "Asia/Seoul", "time.precision.mode": "connect", "transforms": "unwrap", "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState", "transforms.unwrap.drop.tombstones": "false" } } 그래서 debezium으로 mysql의 data를 읽어서 topic으로 가져왔는데ksqldb에서 그 data를 읽는 부분에서 막혔습니다강의에 나온거처럼 ksqldb와 debezium을 연동을 해야 가능한건가요?아니면 어떤 방법이 있을까요?

신규 강의 업데이트 기다리다 너무 화가 나서 글을 올립니다.강사님께 질문드리고 싶습니다 확실히 답변 주세요도대체 강의 업데이트는 언제 됩니까?올해 초부터 강의가 업데이트 된다고 답변하신지 벌써 반년이 다 지나갑니다강사님은 강의를 업데이트 하실 생각이 진정으로 있으십니까? 저야 할인받고 구매하긴 했지만 강의가 한두푼 하는것도 아니고 8만8천원짜리 강의면 취준생들은 정말 고민의 고민 끝에 구매한 금액인데 이렇게 하시면 안되죠 물론 강의를 구입하기 전에 이전버전인걸 알고 구매했으니 제 잘못도 있는게 맞습니다.하지만 강의 이미 녹화 다 끝났다고, 6월 되기 전에 올려주신다고 하셨잖아요저는 강사님의 답변을 믿고 일단 기존 강의 다 들은 뒤에 업데이트 된 강의 들어야지 하는 생각으로 취준생의 입장에서 정말 거금을 들여 구매한건데 정말 화가 나네요. 이렇게 약속을 지키지 않으실걸 알았으면 굳이 이 강의 구매 안하고 타 사이트에 올라온 훨씬 더 저렴한 강의 구매했겠죠 저야 한달정도밖에 기다리지 않았지만 1월부터 기다린 수강생들은 뭐가 됩니까? 강의를 계속할 여력이 되지 않으시면 공지나 질문에 답변하실때 당분간은 좀 어려울 것 같다고 말씀하시거나 현재 진행사항을 솔직히 말씀하셔야하는것 아닌가요?

Spring Boot 3.2 + Spring Cloud 2023.0.0업데이트 예정 (2024-01) 

 tail -f flume-cmf-flume-AGENT-server02.hadoop.com.log 했을때 Creating이나 강의에 말씀한 내용 나오지않고, 아래처럼 나오기만 하는데 Flume Config파일도 정상적이고 재시동도 해봤는데 안되는데 또 조치해야할게 있을까요?

CREATE STREAM add_stream WITH (KAFKA_TOPIC='column_stream_topic', VALUE_FORMAT='JSON') AS SELECT *, CAST(NULL AS INT) AS new_column FROM test_stream EMIT CHANGES;이렇게 기존 test_stream에서 column을 추가한 add_stream을 만들려고 CLI문을 실행시키면원래 test_stream에 담겨있는 data가 담아져서 나오는데package com.example.service; import io.confluent.ksql.api.client.Client; import io.confluent.ksql.api.client.ClientOptions; import io.confluent.ksql.api.client.ExecuteStatementResult; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import javax.annotation.PostConstruct; import java.util.concurrent.ExecutionException; @Service public class streamPracticeAdd { @Value("${ksqldb.server.host}") private String ksqlDbHost; private int ksqlDbPort; private Client client; @PostConstruct public void init() { ClientOptions options = ClientOptions.create() .setHost(ksqlDbHost) .setPort(ksqlDbPort); client = Client.create(options); } public void streamsAdd(String columnName, String dataType) { String createStreamKsql = "CREATE STREAM add_stream WITH (KAFKA_TOPIC='column_stream_topic', VALUE_FORMAT='JSON') AS SELECT *, CAST(NULL AS " + dataType + ") AS " + columnName + " FROM test_stream EMIT CHANGES;"; try { ExecuteStatementResult result = client.executeStatement(createStreamKsql).get(); System.out.println("Stream created and data inserted into new topic: " + result.queryId()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } } }/kafka/addColumn/new_column/INT 인 API 요청을 줘서 새 stream을 만드는 코드인데실행시키면 기존 column에 새 column까지 추가는 되는데 기존 data가 하나도 들어오지 않습니다.검색을 해봤는데도 잘 안나와서 질문 남깁니다 감사합니다.

dependencies에서implementation 'org.springframework.cloud:spring-cloud-starter-gateway-mvc' 부분을 implementation 'org.springframework.cloud:spring-cloud-starter-gateway'으로 바꾸시면 해결됩니다.

컨슈머 리벨런싱 도중에 cpu 과부하 같은 오류가 발생해서 리벨런싱이 완료되지 않은 경우heartbeat.interval.ms 설정은 컨슈머가 안붙었기 때문에 보내지 못하고 session.timeout.ms 설정 값을 통해서 그룹 코디네이터가 컨슈머가 죽었다고 판단해서 리벨런싱이 일어나야 할 것 같은데 --describe 컨슈머 그룹 상태를 확인했을 때 계속 Consumer group '그룹이름' has no active members. 같이 컨슈머가 안붙는다면 서버 재시작을 통해서 강제 리벨런싱을 일어나게 하면 다시 컨슈머가 붙어서 해결이 가능하지만 이 방법을 떠나서 session.timeout.ms 같은 값을 통해서 자동 리벨런싱이 일어나지 않을 수도 있는건지 궁금합니다! 

강의 내용에서는 통합된 키로 암호화와 복호화를 하고 있어, 사실상 대칭 키와 같은 방법이라고 느껴집니다. 강의 내용에서는 비대칭 암호화의 필요성을 못느끼는데 스프링 클라우드 컨피그서버에서 비대칭 암호화에 대한 필요성과 사용하는 이유는 무엇인가요?

안녕하세요. 저번에 카프카 재기동 오류에 관해 질문을 올렸던 수강생입니다. 카프카 실행에 관한 오류는 해결하였습니다만, 이번에는 connector가 실행되지 않는 오류가 발생하여 강사님 말씀대로 카프카 삭제 후 강의와 동일한 버전으로 재설치해보았습니다. 하지만 커넥터는 계속 똑같이 실행되지 않고 있습니다.. 아래는 connect_start_log.sh 입력 후 출력되는 로그 내용입니다.... 이 뒤로는 Retrying to fetch metadata만 반복적으로 출력됩니다. 끝없는 오류의 굴레에서 벗어나고 싶습니다.. 제발 도와주세요