깃허브 코드대로 했습니다

안녕하세요. IoT Device들의 데이터를 받아서 값들이 조건을 만족하면 이벤트를 발생시키는 개발업무에 있습니다. 그래서 streaming 분석솔루션을 찾던도중 ksqlDB관련을 접하게되었습니다. 강의를 간단하게 훑어보고 ksqlDB 솔루션을 적용해보고싶은데 궁금한점이있어서 질문 드립니다. IoT Device의 개수가 많을 경우에 각각의 Device의 걸어주어야하는 조건이 다양해질수있습니다.만약에,A device의 조건은 해당 값이 10이 넘어가면 이벤트 발생 그리고 B device의 조건은 값이 15가 넘어가면 이벤트발생 이런식으로 구성될수있습니다.각 Device마다 고유한 조건이 있기때문에 많은 stream이 생성될수있습니다.또한 E Device에서 발생한 이벤트 와 J Device에서 발생한 이벤트를 조인해서 새로운 stream을 만들어주어야할때도있습니다. 디바이스가 많고 이벤트처리 흐름이 복잡해질수록 stream,table,Mview가 많이 생길수 있기때문에 성능에대한 우려가 있습니다.물론 H/W나 traffic, SQL문의 복잡도,네트워크구성등 성능에 미치는 요인들이 많기때문에 성능이 어떻다는 확실히 답변하기 어려울것같긴합니다. 결론은 데이터스트리밍 분석을 위해 많은 stream,table,Mview 작성해도 성능 이슈가 크게 발생하는지(몇백 혹은 몇천개단위). 만약 성능이 부족하다면 kafka를 scale out,up하면 나아질수있는지도 궁금합니다. 강의를 전체적으로 훑는식으로봐서 자세히 듣지못해 틀린부분이 있을수있지만, 부족한점 참고하여 봐주시면 감사하겠습니다.   

config를 root dir에 위치시킨게 아니라 특정 dir에 넣고 싶은데 그럴 때는 어떻게 uri를 잡아야하나요?

안녕하세요. 강의 잘 듣고있습니다 !카프카 컨슈머가 쿠폰레포지토리에 save하는 과정에서 모종의 이유로 실패하여 롤백 되었을 때, 레디스에서는 이미 increment가 되었는데 이걸 하나 빼줘야되는거 아닌가요?이런 처리는 어떻게하는지 궁금합니다 !추가) 뒤에 얘기가 나오는군요 ㅎㅎㅎ.. FailedEvent를 저장하는 것도 실패하였다면, 배치프로그램이 제 역할을 못할 것 같은데요.. ! 그렇다면 로그를 보고 판단해야할 것 같은데 실무에서도 그렇게 진행이 될까요 ? 그리고, FailedEvent를 삽입하는 것에서도 에러가 발생할 수 있기 때문에 또 try catch로 묶어서 이것 마저 실패할 경우에도 로그를 따로 남겨야하는지 궁금합니다 !

jjwt가 버전업 되면서 강사님이 업데이트 해주신 코드들도 jwt valid할 때 안되는 경우가 발생하더라고요... 그래서 찾아본결과다음과 같이 작성하시면 원활하게 동작을 합니다. 참고하세요.  private boolean isJwtValid(String jwt) { byte[] secretKeyBytes = Base64.getEncoder().encode(env.getProperty("token.secret").getBytes()); SecretKey signingKey = new SecretKeySpec(secretKeyBytes, SignatureAlgorithm.HS512.getJcaName()); boolean returnValue = true; String subject = null; try { JwtParser jwtParser = Jwts.parser(). verifyWith(signingKey).build(); Jws<Claims> claimsJws = jwtParser.parseSignedClaims(jwt); Claims payload = claimsJws.getPayload(); subject = payload.getSubject(); }catch (Exception ex){ returnValue = false; } if(subject == null || subject.isEmpty()){ returnValue = false; } return returnValue; }

안녕하세요.. 수강 하면서 하다 보니, 본의 아니게 수정 사항이 보입니다.KSQLDB/Join 이해 중에 select절에서 a.id 가 아닌 , a.user_id 로 수정 해야 될거 같습니다.

zookeeper에서 오류가 발생해서 테스트 진행시 어려움이있습니다. 강의상의 내용으로 진행되었고,해당내용으로는 진행되지 않아 웹서핑 이후version: '3'services: zookeeper: image: zookeeper ports: - "2181:2181" kafka: build: . ports: - "9092:9092" environment: KAFKA_ADVERTISED_HOST_NAME: localhost KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 volumes: - /var/run/docker.sock:/var/run/docker.sock확인되는 내용으로 변경했는데도 잘 안되더라구요 ㅜ늦게라도 확인되신다면 회신부탁드립니다!!

안녕하세요,문득 실무에서 카프카 클러스터 환경 구축에 대해 궁금한 점이 생겼습니다. 강사님께서 진행하신 방법은 confluent의 카프카 프로젝트를 하나의 노드에 직접 다운받아서 주키퍼, 브로커 기동도 하고 ksql-cli 실행도 하신 것 같은데요,실무 환경에서는 클러스터 구축 노드가 주키퍼 클러스터를 이루는 노드들 및 브로커 등 여러 가지 일텐데요, 이런 경우는 어떻게 진행하는게 일반적일까요? 강의에서 사용한 동일한 플랫폼용 카프카 프로젝트를 ansible 등으로 설치 지원이 되는지 궁금합니다. 추가로 주키퍼나 브로커 노드들은 그 기능만 수행하고 ksql-cli나 schema-registry 혹은 connect 용으로 노드를 별도로 두고 사용하는 것이 일반적인지 궁금합니다.  

안녕하세요. 강좌를 수강 하다가, 강의 내용중 Mview CSAS 에서, customer_activeity_stream insert 문장이 강의 내용과 다르고, 칼럼이 달라서 에러 발생 합니다. activety_id 없습니다.  감사 합니다. INSERT INTO customer_activity_stream (customer_id, activity_id, activity_type, activity_point) VALUES (2, 10,'mobile_open',0.65); INSERT INTO customer_activity_stream (customer_id, activity_id, activity_type, activity_point) VALUES (4, 3, 'deposit', 0.35);

하나의 토픽, 하나의 파티션으로 순서보장을 하는 Consumer 처리를 알고싶습니다.kafka paraller-consumer 처리를 하고 싶은데 마땅한 예제가 나와있지 않아서요.해당 예제를 찾아보면 보통 multi thread 방식이라 순서보장이 되지않은것 같아 사용을 못할것 같습니다.최종적으로 순서보장을 하는 consumer 병렬처리 방법을 알고 싶어 질문드렸습니다.

안녕하세요."섹션 16. 애플리케이션 배포 - Docker Container"을 학습하면서 config-service를 Docker로 만들어 실행하면 컨테이너 실행 됐다가 Exit가 되고, "docker logs config-service"로 로그를 보면 아래와 에러가 발생합니다. JDK11 , springboot 2.4.5 버전에 spring-cloud-starter-bootstrap 의존성은 추가 되어 있는 상태입니다. 답변 부탁드립니다. 감사합니다.===================================================================2024-03-22 06:50:22.104 WARN 1 --- [ main] ConfigServletWebServerApplicationContext : Exception encountered during context initialization - cancelling refresh attempt: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'healthContributorRegistry' defined in class path resource [org/springframework/boot/actuate/autoconfigure/health/HealthEndpointConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.boot.actuate.health.HealthContributorRegistry]: Factory method 'healthContributorRegistry' threw exception; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'configServerHealthIndicator' defined in class path resource [org/springframework/cloud/config/server/config/EnvironmentRepositoryConfiguration.class]: Unsatisfied dependency expressed through method 'configServerHealthIndicator' parameter 0; nested exception is org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'org.springframework.cloud.config.server.config.CompositeConfiguration': Unsatisfied dependency expressed through method 'setEnvironmentRepos' parameter 0; nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'defaultEnvironmentRepository' defined in class path resource [org/springframework/cloud/config/server/config/DefaultRepositoryConfiguration.class]: Invocation of init method failed; nested exception is java.lang.IllegalStateException: You need to configure a uri for the git repository.2024-03-22 06:50:22.113 INFO 1 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'applicationTaskExecutor'2024-03-22 06:50:22.117 INFO 1 --- [ main] o.apache.catalina.core.StandardService : Stopping service [Tomcat]2024-03-22 06:50:22.134 INFO 1 --- [ main] ConditionEvaluationReportLoggingListener :Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.2024-03-22 06:50:22.149 ERROR 1 --- [ main] o.s.b.d.LoggingFailureAnalysisReporter :***************************APPLICATION FAILED TO START***************************Description:Invalid config server configuration.Action:If you are using the git profile, you need to set a Git URI in your configuration. If you are using a native profile and have spring.cloud.config.server.bootstrap=true, you need to use a composite configuration.=================================================================== 

안녕하세요. 강의 잘 보고 있습니다. 저는 현재 kotlin, postgreSQL을 사용하고 있으며M3 맥북 환경입니다 강의에서 쓰레드슬립 사용 전 테스트케이스가 실패하는걸 확인할 수 있는데, 왜 저는 테스트가 성공되는걸까요..?? 원래 실패 해야 정상인 상황일까요? 어떤 이유에서 성공하는지 감이 잘 안옵니다..

안녕하세요https://www.inflearn.com/questions/1207899/%EC%8A%A4%ED%82%A4%EB%A7%88-%ED%98%B8%ED%99%98%EC%84%B1-%EC%A7%88%EB%AC%B8%EC%9E%88%EC%8A%B5%EB%8B%88%EB%8B%A4이 질문에 대한 추가 질문이 있습니다. BACKWARD에 대해 지원하는 신규 스키마 변경에서기존 필드 삭제, 기본 값 가진 신규 필드 추가이 부분은 sink쪽 스키마에서의 변경에 대한 내용인 것은 맞는건가요? 예를 들어 스키마 호환성 체크에서새로 추가되는 컬럼이 기본 값을 가지고 있지 않으면 하위 호환성 오류다. 라는 문장이 있다면"sink 쪽에서" 새로 추가되는 컬럼이 기본 값을 가지고 있지 않으면 하위 호환성 오류다. 라고 해석하는 것이 맞는지 질문드립니다.

안녕하세요.해당 수업을 잘듣고 있습니다.혹시 consumer에 대한 병렬처리방법에 대한 강의가 있을까요?참고할만한 강의가 있을까요?감사합니다.

안녕하세요, schema registry를 사용할 때 avro를 사용해야한다고 말씀해주셨는데요, Json 포맷은 지원하지 않는건가요? 물론 avro의 경우 필드 정보가 payload에 빠지고 바이너리 형태이니 전송이 가장 빠르고 최적화된 포맷이라는 것은 알지만, Json 포맷도 설정을 통해 스키마를 지원할 수 있도록 할 수 있는데 schema registry의 지원 대상에서 제외되는 것인지 궁금합니다.

replication lag이라는게 존재하면, 데이터가 리더 파티션에 들어올 때 마다 약간의 지연 시간이 팔로워에서 발생한다는 건데, 그말인 즉슨 ISR이 거의 유지될 수 없다처럼 느껴져서요.. 이부분이 궁금합니다.

안녕하세요, 스태틱 그룹 멤버십 관련 질문있습니다!만약 group instance id를 동일 counsumer group에 속한 counsumer 중에 일부에만 할당하는 경우는 어떻게 되나요?이 경우에도 스태틱 그룹 멤버십이 적용되는 건가요?

안녕하세요. 카프카가 데이터를 삭제할 때 세그먼트, 즉 파일 단위로 삭제한다고 말씀해주셨습니다.retention.ms는 active 세그먼트가 아닌 세그먼트가 해당 기간을 지났을 때 삭제한다는 것이 명료해서 이해가 갔는데, retention.byte는 조금 헷갈립니다.세그먼트의 기본 크기가 1GB로 알고 있습니다. 만약 retention.ms를 사용하지 않는다고 가정하면 (매우 큰값이라고 가정한다면) 세그먼트 삭제는 retention.byte에 의해 좌우될텐데요, retention.byte를 1GB보다 크게 주는 경우, 세그먼트는 영영 삭제되지 않는 것일까요?

안녕하세요, postgresql의 logical replication에서는 publication이라고하는 모듈을 통해 복제 대상 테이블을 정해놓는데요 이렇게 되면 WAL에는 대상 테이블의 변경 내용만 담기는 건가요? 그리고 mysql에는 publication과 같은 모듈이 없는 것 같은데 bin 로그에는 그럼 모든 테이블의 트랜잭션 로그가 담기는 건지 궁금합니다.  

안녕하세요, 디비지움 소스 테이블 스키마 변경에 따른 자동 스키마 반영에 대한 질문이 있습니다. 소스 테이블에 변경이 있을 때 예를 들어 varchar 타입 컬럼에 default를 적용하는 변경을 수행하는 경우 sink 쪽에서 오류가 발생하는 것으로 알고 있습니다.그런데 만약 디비지움에 스키마 레지스트리가 적용되어 있고, 스키마 호환성을 체크한다고 한다면디비지움의 소스 테이블 변경 규칙은 무시될수도 있으며, 스키마 레지스트리 호환성에 따라 동작하게 되는 것인가요?