안녕하세요 JDBC Sink connector에 config 옵션으로 pk.mode를 통해 record key 값을 지정해주어야 하니, source connector 쪽에서도 transform을 통해 record key 값으로 pk를 추출해야 한다는 명분은 이해했는데요, value에 pk 값이 이미 있고, config를 통해 pk 필드가 무엇인지 까지 지정해줬는데 value를 통해 값을 획득하지 않고 record key에 다시 추출해야하는 이유가 뭘지 궁금합니다!

안녕하세요.친절히 잘 알려주셔서, 감사히 강의를 잘 듣고 있습니다.또, 강의를 듣고, 매일 복습을 하며, 바쁜 날을 보내고 있습니다.맨 처음에 CentOS를 설치하고, VM를 세팅하는 과정에서 호스트 전용 네트워크 정보에 192.168.56.xxx와 같은 로컬 네트워크 IP를 입력하였던 것으로 기억합니다.제가 문의 드리고 싶은 부분은 만약 PC-A와 PC-B가 있을 때,PC-A에 server01, server02를 구축하고, PC-B에서 putty를 이용해서 원격 접근을 하려고 할 때, 단순히 로컬 네트워크 IP를 외부 네트워크 IP로만 변경해서 사용하면 되는 것인지, 만약, 그렇지 않다면, 추가적으로 어떤 설정을 해야하는지 여쭙고 싶습니다.이와 관련된 자료의 링크를 남겨주시면 감사하겠습니다.또한, 두번째 질문으로, 클러스터란 다수의 컴퓨터 서버를 모아 놓은 것으로 알고 있습니다.저는 현재 저사양 컴퓨터 기준으로, server01, server02를 Cluster-1로 제어하고 있는데요. 실제 실무에서도, 다수의 server를 오직 하나의 Cluster로만 그룹하여 사용하는지, 그게 아니라면, Cluster로 그룹하는 기준은 무엇인지 궁금합니다.감사합니다.

안녕하세요 ! 강사님좋은 강의 잘 보고 있습니다.Feign의 예외처리에서 강의와 다른부분이 발생하여 질문 남깁니다. FeignErrorDecoder를 통해 예외 처리시 .[user-service] [-auto-1-exec-10] .w.s.m.a.ResponseStatusExceptionResolver : Resolved [org.springframework.web.server.ResponseStatusException: 404 NOT_FOUND "사용자의 주문이 비었습니다"] 위 로그가 user-service에서 error catch를 통해 잘 나오는 것을 확인했습니다.(09:36)강의영상에서는 404 Error를 전달하여 포스트맨 응답에서 Error msg까지 출력이 되었지만, 저의 경우에선 APIGatgeway에서의 로그및 포스트맨응답에Global Filter End : response id -> 403 FORBIDDEN으로 응답이 보내집니다. 이전 강의 부분(정상 응답)에서 해당 문제가 없었지만, 에러 응답시 403응답으로 처리되어지는 부분에 대해서 추가적으로 ApiGateWay에 대해서 github에서는 따로 handling이 없어서 추가적으로 handler를 등록하여 적용되어야하는 부분인지 여쭤봅니다 .++해당부분 user-service의 WebSecurity특정 ip 통과부분에 실제 ip(자신의 ip)를 누락하였네요.Error에 대한 부분도 응답이 됩니다.추가적으로 error에 대해서 403으로 응답이 되는부분은https://colabear754.tistory.com/182해당글을 참고하시면 될것같습니다.  

gateway에 의해서 api는 정상적으로 호출이 되는데 로그가 출력되지 않습니다.customFilter, globalFilter 모두 출력이 되지 않았습니다. 코드에 혹시 누락된 부분이 있을까요.Global Filterpackage com.example.springgateway.filter; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.springframework.cloud.gateway.filter.GatewayFilter; import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.stereotype.Component; import reactor.core.publisher.Mono; @Component @Slf4j public class GlobalFilter extends AbstractGatewayFilterFactory<GlobalFilter.Config> { public GlobalFilter() { super(Config.class); } @Override public GatewayFilter apply(Config config) { // Custom Pre Filter return (exchange, chain) -> { ServerHttpRequest request = exchange.getRequest(); ServerHttpResponse response = exchange.getResponse(); log.info("Global Filter baseMessage : {}", config.getBaesMessage()); if (config.isPreLogger()) { log.info("Global Filter Start : request id -> {}", request.getId()); } else { log.info("preLogger is false"); } // Custom Post Filter return chain.filter(exchange).then(Mono.fromRunnable(()-> { if (config.isPostLogger()) { log.info("Global Filter End : request id -> {}", response.getStatusCode()); } else { log.info("postLogger is false"); } })); }; } @Data public static class Config { //put the configuration properties private String baesMessage; private boolean preLogger; private boolean postLogger; } } application.ymlspring: application: name: gateway-service cloud: gateway: routes: - id: first-service uri: http://localhost:8081/ predicates: - Path=/first-service/** filters: - CustomFilter - id: second-service uri: http://localhost:8082/ predicates: - Path=/second-service/** filters: - CustomFilter # - AddRequestHeader=second-request, second-request-header2 # - AddResponseHeader=second-response, second-response-header2 - id: user-service uri: http://localhost:8083/ predicates: - Path=/user-service/** filters: - CustomFilter # filters: # - AddRequestHeader=user-request, user-request-header2 # - AddResponseHeader=user-response, user-response-header2 - id: catalog-service uri: lb://CATALOG-SERVICE predicates: - Path=/catalog-service/** - id: order-service uri: lb://ORDER-SERVICE predicates: - Path=/order-service/** default-filters: - name: GlobalFilter args: baseMessage: Spring Cloud Gateway Global Filter preLogger: true postLogger: true eureka: client: register-with-eureka: true fetch-registry: true service-url: defaultZone: http://localhost:8761/eureka server: port: 8000 Filter Configpackage com.example.springgateway.config; import org.springframework.cloud.gateway.route.RouteLocator; import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; //@Configuration public class FilterConfig { // @Bean public RouteLocator gatewayRoutes(RouteLocatorBuilder builder) { return builder.routes() .route(r -> r.path("/user-service/**") .filters(f -> f.addRequestHeader("user-request", "user-request-header") .addResponseHeader("user-response", "user-response-header")) .uri("http://localhost:8083/")) .route(r -> r.path("/second-service/**") .filters(f -> f.addRequestHeader("second-request", "second-request-header") .addResponseHeader("second-response", "second-response-header")) .uri("http://localhost:8082/")) .build(); } } Postman 실행 결과

안녕하세요  "4.실시간 적재 파일럿 실행 2단계 03 - Redis 설치 " 강의 중, gcc와 tcl 설치 도중에 에러가 발생해서 질문 드립니다. $ yum install -y gcc*해당 에러가 url 변경으로, 에러가 발생해서 echo "https://vault.centos.org/6.10/os/x86_64/" > /var/cache/yum/x86_64/6/base/mirrorlist.txt echo "http://vault.centos.org/6.10/extras/x86_64/" > /var/cache/yum/x86_64/6/extras/mirrorlist.txt echo "http://vault.centos.org/6.10/updates/x86_64/" > /var/cache/yum/x86_64/6/updates/mirrorlist.txt 로 변경하여 설치를 시도했습니다.하지만 아래 사진과 같이 여전히 gcc가 설치되지 않음을 확인하였습니다. 그리고, tcl에 대해서도 설치 과정에서 아래 사진과 같이 에러가 발생했습니다. 이 문제에 대한 해결 방법을 알려주시면 감사하겠습니다.위 사진에 대한 내용도 아래에 작성해놓겠습니다. [root@server02 ~]# echo "https://vault.centos.org/6.10/os/x86_64/" > /var/cache/yum/x86_64/6/base/mirrorlist.txt [root@server02 ~]# echo "http://vault.centos.org/6.10/extras/x86_64/" > /var/cache/yum/x86_64/6/extras/mirrorlist.txt [root@server02 ~]# echo "http://vault.centos.org/6.10/updates/x86_64/" > /var/cache/yum/x86_64/6/updates/mirrorlist.txt [root@server02 ~]# rpm -q gcc package gcc is not installed [root@server02 ~]# [root@server02 ~]# [root@server02 ~]# yum install -y tcl Loaded plugins: fastestmirror, refresh-packagekit, security Setting up Install Process Loading mirror speeds from cached hostfile https://archive.cloudera.com/cm6/6.3.1/redhat6/yum/repodata/repomd.xml: [Errno 14] PYCURL ERROR 22 - "The requested URL returned error: 404 Not Found" Trying other mirror. To address this issue please refer to the below knowledge base article https://access.redhat.com/articles/1320623 If above article doesn't help to resolve this issue please open a ticket with Red Hat Support. Error: Cannot retrieve repository metadata (repomd.xml) for repository: cloudera-manager. Please verify its path and try again [root@server02 ~]# [root@server02 ~]# [root@server02 ~]#  감사합니다.

안녕하세요, 강의에서 connector plugin.path 지정 시 직접 생성한 dir에 plugin 별 서브 dir를 두고 jar 파일을 옮겼는데요, 서브 디렉토리가 필요한 이유가 있을까요? 그리고 서브 디렉토리명은 임의로 지어도 되는 것인지 궁금합니다! 감사합니다.

안녕하세요"6. 적재 파일럿 실행 4단계 - 적재 기능 테스트" 강의 중, 플럼 이벤트 작동과 관련되어 실습 중 질문이 있어 문의드립니다.현재 /home/pilot-pjt/working/SmartCar 경로에 마들어진 SmartCarStatusInfo_20160101.txt 파일을 플럼의 SmartCarInfo의 SpoolDir 경로인 /home/pilot-pjt/working/car-batch-log로 옮겨 플럼의 File 이벤트가 작동하기 위해, $ mv /home/pilot-pjt/working/SmartCar/SmartCarStatusInfo_20160101.txt /home/pilot-pjt/working/car-batch-log/$ cd /var/log/flume-ng/$ tail -f /var/log/flume-ng/flume-cmf-flume-AGENT-server02.haddop.com.log를 입력하였습니다. 이 후, "...BuckerWriter: Closing /pilot-pjt/...", "...BuckerWriter: Renaming /pilot-pjt/...", 그리고 마지막으로 "...Writer callback called"라는 메세지가 보이면 모든 HDFS 적재가 성공적으로 끝난 것이다라고 설명해주셨지만, 저는 아래 출력값으로 약 40분 간 추가적인 출력이 발생하지 않은 상황입니다.[root@server02 SmartCar]# cd /var/log/flume-ng/ [root@server02 flume-ng]# tail -f /var/log/flume-ng/flume-cmf-flume-AGENT-server 02.hadoop.com.log value.serializer = class org.apache.kafka.common.serialization.ByteArray Serializer 2024-01-28 01:11:50,605 INFO org.eclipse.jetty.server.Server: jetty-9.3.25.v2018 0904, build timestamp: 2018-09-05T06:11:46+09:00, git hash: 3ce520221d0240229c86 2b122d2b06c12a625732 2024-01-28 01:11:50,743 INFO org.eclipse.jetty.server.AbstractConnector: Started ServerConnector@553dd1e0{HTTP/1.1,[http/1.1]}{0.0.0.0:41414} 2024-01-28 01:11:50,743 INFO org.eclipse.jetty.server.Server: Started @2563ms 2024-01-28 01:11:51,114 INFO org.apache.kafka.common.utils.AppInfoParser: Kafka version: 2.2.1-cdh6.3.2 2024-01-28 01:11:51,117 INFO org.apache.kafka.common.utils.AppInfoParser: Kafka commitId: null 2024-01-28 01:11:51,119 INFO org.apache.flume.instrumentation.MonitoredCounterGr oup: Monitored counter group for type: SINK, name: DriverCarInfo_KafkaSink: Succ essfully registered new MBean. 2024-01-28 01:11:51,121 INFO org.apache.flume.instrumentation.MonitoredCounterGr oup: Component type: SINK, name: DriverCarInfo_KafkaSink started 2024-01-28 01:11:51,161 INFO org.apache.kafka.clients.Metadata: Cluster ID: lJYz nSt_QWWTqgLn1n7JwA   이런 상황에서 원인이 무엇인지와 해결 방법에 대해 알려주시면 감사하겠습니다. 항상 강의를 통해 많은 것을 공부하고 있습니다.감사합니다. 

궁금힌ㅂ니다

안녕하세요 우선 좋은 강의 감사드립니다.다름이 아니라 강의에서 Connect Task 가 thread 로 작동한다고 말씀하신 부분을한 Worker process 내에서 여러 개의 thread(task) 로 병렬 처리를 할 수 있다 라고 이해했습니다. 그럼 혹시 여러 Task 가 동시에 접근할 수 있는 변수를 선언해서 사용할 수 있을까요? race_condition 같은 문제를 해결하기 위해 mutex 를 사용할 것 같긴 한데 이러한 구현이 가능한 지 궁금합니다. 감사합니다

안녕하세요. mac m1에 utm으로 고정IP(192.168.56.101) 설정후 ssh 접속 및 kafka-console 명령어 잘 됩니다.하지만, java 코드로 실행시 아래 이미지와 같이 접속 이슈가 있어서 문의 드립니다.ubuntu 설정에 이름이 "ubuntu"로 설정하게 문제일까요?

안녕하세요,  confluent local도 bin 스크립트를 보니 kafka_server_start가 있고 이를 통해 커뮤니티처럼 멀티 브로커 클러스터를 생성할 수 있는 것이 아닌지 궁금합니다. 또한 실제로 confluent_local을 통해 모니터링 하는 사례가 현업에서 있는지 궁금합니다.

안녕하세요, log dir에 대해 질문이 있습니다.제가 실습을 하다보니 kafka-logs-0? 디렉토리에 많은 하위 디렉토리들이 생겨서 새로운 실습을 위해 비웠습니다. 기동 중인 모든 브로커에 대한 log dir를 초기화하였는데, 브로커에서 log dir관련 에러로그가 발생한 후 shutdown이 되길래 다시 구동을 시켰습니다.  그리고 각 브로커의 log dir를 다시 확인하니 topic 파티션 dir를 비롯해서 기존 모든 dir가 복원되어있었습니다. 제가 궁금한 점은 카프카 클러스터가 동작하면서 브로커만 있는 것이 아니니 복원 지점이 어디엔가 있을 수 있겠다고 생각은 하였는데, topic-partition의 log들은 replication을 배울 때 혹시 특정 브로커(노드)에 문제가 생길 때를 위한 복제라고 배운 것 같은데, 다른 모든 브로커도 모두 이 정보가 지워진 상태에서 어떻게 복원이 가능했던 걸까요?   

안녕하세요, 동일 groupId를 가진 두 개의 컨슈머가 하나의 토픽에 대해 poll을 하는데, 컨슈머 중 하나는 0번 파티션을 읽도록 assign하였고 나머지 컨슈머는 그냥 토픽에 대해 subscribe를 하였습니다. 그런데 0번 파티션에 assign한 컨슈머는 데이터를 읽어오지 못하고 아래와 같은 에러 로그가 발생하였는데요 Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member 이 경우는 어떻게 해석하는게 좋을까요? 

해당 예제를 통해 주키퍼를 먼저 실행을 해볼려하지만 계속해서 해당 classpath를 찾지 못하고 있습니다. 비슷한 상황을 겪은 사람들의 경우 바이너리 파일이 아닌 소스 파일을 다운 받아 해당 문제를 겪었다고 했지만 저의 경우 알맞은 파일을 설치하여 압축을 해제하였는데 어떤게 원인일까요

안녕하세요. confluent kafka 에서 consumer 뛰운 상태 입니다.SimpleProducer.java 작성 후, Run 하면 아래 로그 발생 합니다.SLF4J(W): No SLF4J providers were found. consumer에서 메시지 출력 하지 않습니다. 검색해 보니, This warning, i.e. not an error, message is reported when no SLF4J providers could be found on the class path. Placing one (and only one) of the many available providers such as slf4j-nop.jar slf4j-simple.jar, slf4j-reload4j.jar, slf4j-jdk14.jar or logback-classic.jar on the class path should solve the problem. 어디를 수정 해서, 적용 해야 할지 잘 모르겠습니다. 강의 내용 따라서, Gradle 설정 하고, jdk를 확인 해보니, 17 oracle jdk 가 보이지 않아, jbrskd-17로 설정해서 빌드 했습니다. (제 intellij에서 add > oracle 17jdk가 안 보입니다???) 검토 하시고 답변 부탁 드립니다.    

좋은강의 잘듣고있습니다! 다름아니라 현재 싱크 커넥터등록후 DB에 데이터가 등록되지 않고 있습니다.먼저 아래사진처럼 post요청을 통해 커넥터 등록을 하였습니다.토픽은 예제를 똑같이 따라하였을 때도 안되었어서 my-topic-orders로 새로 바꿔서 등록하였습니다.이에 맞게 테이블도 아래 사진처럼 생성하였습니다. post요청을 한 후 주문이 해당 토픽으로 전달했음에도 아래 사진처럼 DB에 적용이 되지않아 어디가 문제일지 커뮤니티를 뒤져도 잘모르겠어서 질문드립니다 ㅠㅠ (토픽도 "my_topic_orders"로 지정하였습니다.)현재 커넥터 등록상태입니다.추가로 kafka confluent에서 발생한 로그입니다

안녕하세요, 프로듀서 전송 부분에 대해 강의를 듣던 중 궁금한 점이 있습니다.최대 한 번 전송 외에는 ack를 받고 다음 메시지를 전송한다고 설명하셨는데, 비동기 전송은 ack와 상관 없이 지속적으로 메시지를 전송할 수 있는거 아닌가요? 메시지 a,b,c 3개를 보낸다고 했을 때ack = 0의 경우는a,b,c를 보내고 오는 ack에 대해 신경 쓰지 않는것이고,ack != 0의 경우 비동기라면 마찬가지로 전송은 ack 여부와 상관 없이 a,b,c를 보내고 대신 ack 여부에 따라 특정 메시지에 대한 재전송이 이루어 지는 것이 아닌지 궁금합니다.즉, ack 여부에 따라 a이후 b를 전송하는 것이 결정되는게 맞는지 궁금합니다.       

안녕하세요. 저는 1년차 개발자 입니다.현재는 회사를 그만 두고 공부를 마치고 취업 준비를 하고 있습니다. 권철민 개발자님의 강의를 접하고 너무 좋아서,kafka 3개의 모두 수강하고 마지막 강의를 듣고 있습니다.지식을 모두에거 공유한다는 것이 너무 존경스럽고 개발자가 가져야하는 자세와 덕목이라 생각하고 배우게 되었습니다. 감사합니다. 실무에서 실제로 사용 하고 싶어서 집에서 AWS에 환경을 구축 해보고 싶습니다. 하지만 비용이 청구 된다는 것에 두려움을 가지고 있습니다. AWS는 실제로 한번도 없지만 클라우드 환경에 대한 선수 지식과 AWS사용 법등은 숙지하고 있지만, 비용적인 측면에서 기술을 사용하는데 있어 비용이 어떤 정책으로 측정 되는지 알고 있는게 없어 너무 걱정입니다. confluent kafka 와 apache kafka 사용법에 차이는 별로 없다고 하셧지만, 그래도 배운 confluent kafka로 aws를 구축 하고 싶은 마음이지만 몇가지 궁금 한게 있습니다. 현재 현업에서는 confluent kafka 와 apache kafka 둘 중 어떤것이 많이 사용되며, 선호 되는 것이 어떤건가요?현업에서는 실제로 구축한다면, 강의에서와 같이 Ubuntu AWS에서 만들어서,kafka를 설치하여 사용 하나요? 아니면 제공 되어지는 Saas를 통해 사용하나요? 보통 어떤것을 사용하나요? 아니면 다른 방법이 있나요? 실무에서 사용 하려면 kafka가 무료가 아니라고 알고 있습니다. confluent kafka 와 apache kafka 둘다 무료가 아닌건가요? AWS사용 비용외 별도 비용을 지불 하고 사용 해야 하는 건가요?, 아니면 Saas를 사용하면 AWS비용을 포함하여 비용이 청구 되나요? confluent kafka 와 apache kafka 둘다 똑같은 환경을 구축 한다면, 비용적 측면에서 어떤것이 저렴 한가요? 아니면 걱정 할 정도의 비용 차이가 나지 않는 건가요? 실제 현업에서 어느 정도 규모에서 사용 되며, 작은 규모, 중간 규모, 큰 규묘 에서 대략 비용이 대략 한달 얼마 정도 청구 되나요? 급하게 두서 없이 작성한거 같네요. 긴 글 읽어 주셔서 감사합니다.    

안녕하세요, 인텔리제이에서 클라이언트 프로듀서 생성 후 데이터 전송 시 브로커를 찾을 수 없다는 에러 로그가 자꾸 발생해서, server.properties의 advertised.listeners에서 저희가 사용하는 고정 IP를 직접 작성해주니 해결되었는데요, 이렇게 하니 외부 클라이언트에서 192.168.56.101 주소로 전송한 데이터는 VM에서 콘솔 컨슈머를 통해 read 할 때 localhost로 접근 시 조회가 되지 않습니다. 혹시 Host PC에서 VM의 브로커에 send 할 때 advertised.listeners 값을 고정 IP로 등록해주어야 하는 이유가 있는지, 그리고 이렇게 등록 시 VM 내부에서도 localhost 참조가 불가능해지는게 맞는지 알 수 있을까요?

안녕하세요, producer의 send 로직을 보면 Serialization을 먼저 수행하고 Partitioner에서 파티션을 결정 짓는것으로 보이는데요,key 값이 존재하는 경우 key 값에 대한 hash 알고리즘을 통해 파티셔너에서 파티션을 결정한다고 하였는데 직렬화 되어버리면 hash는 bytes 값에 대한 hash가 수행되는 건가요?? 만약 맞다면, 파티셔너도 특정 키를 가진 데이터에 대해서 원하는 파티션을 지정하도록 커스터마이징이 가능한 것으로 알고 있는데, 파티셔너를 커스터마이징한다면 key 값을 직렬화 하고 그것에 대한 해시알고리즘을 적용해서 파티션 넘버를 반환하도록 짜야하는 것인지 궁금합니다.