안녕하세요, 먼저 좋은 강의 잘 듣고 있습니다 (- -) (_ _) 컨슈머 랙 모니터링 - 처리량 이슈 파트에서 궁금한 점이 생겨서 질문드립니다. 파티션 1개 - 컨슈머 1개에서 파티션 2개 - 컨슈머 2개가 되면 linear하게 처리량이 늘어난다고 하셨는데, 어떻게 처리량이 늘어나는 건지 조금 이해가 안가서요. 상단과 같은 구조는 하나의 컨슈머가 0,1,2,3... 이렇게 하나씩 처리를 한다고 하면, 하단과 같은 구조는 다음과 같이 레코드를 동시에 소비하게 되어 2배로 늘어난다는 말씀일까요? 컨슈머 1 - 0 : 2 : 4 : 6 : ... 컨슈머 2 - 1 : 3 : 5 : 7 : ... ( : 는 레코드가 소비되는 시점을 구분한 것 입니다!) 만약 그렇다면, offset은 어떻게 관리가 되는건지도 궁금하네요..
강사님, 안녕하세요! 강의 잘 듣고 있습니다. 강의 들으면서 궁금한 사항이 생겨서 질문드려요. key를 지정해서 메시지를 전송하면 같은 key에 한해서는 같은 파티션으로 메시지가 전송되는 것을 보장 받을 수 있고, 해당 key로 전송된 메세지는 순서를 보장받을 수 있다. 까지는 이해했는데요~ 운영하면서 파티션의 개수가 늘어날 경우, 늘어난 직후에도 같은 key로 전송되는 메세지는 순서를 보장받을 수 있는 것일까요? (즉, 파티션 개수가 늘어나기 이전에 key:a, partition:0으로 전송되었다면 추후, 파티션 개수가 늘어나 리밸런싱이 일어나더라도 key:a 메세지는 partition:0으로 전송되는 것일까요?)
KTable이 로컬 기반의 rocksDB에 기록된 데이터를 가져다 쓴다는 설명을 듣고 궁금증이 생깁니다. 카프카 스트림을 docker기반 컨테이너로 배포한다 치면 그때마다 rocksDB는 초기화 될 것으로 생각되는데요.. 그렇다면 조인시에 데이터가 없어서 의도한 동작이 이루어지지 않을 수 있겠다는 생각이 드는데 맞는지요? 마찬가지로 windowing 프로세스도 집계중에 컨테이너가 갈아쳐지면 집계중이던 데이터는 소실될 수 있어보이는데 맞는지 모르겠습니다. 실무에서는 보통 카프카 스트림즈를 어떻게 쓰는지가 궁금합니다. rocksDB를 따로 두어서 모든 스트림이 공유하도록 설정하거나 컨테이너 배포는 사용하지 않는지.. 아니면 EC2같은곳에 배포해서 로컬 DB데이터를 훼손시키지 않도록 하는지 같은것이요. 감사합니다.
만약 KStream, KTable 파티션 개수가 2개이고, 파티셔너 전략도 동일합니다. 근데 데이터 발생양이 증가하여, 파티션 개수를 둘다 5개를 늘려야 하는 상황이 생겼습니다. 이때는 어떻게 해야하나요? 하나 씩 파티션을 증가할 때, 파티션 개수가 다르면 TopologyException이 발생할텐데요. 또, 파티션이 추가되면 파티션 1번으로 가던 메시지 키가 다시 1번으로 간다는 보장도 없고요.. 2개의 토픽을 리파티셔닝 작업을 해야하는걸까요? 리파티셔닝 작업을 하는 동안은 스트림즈의 다운 타임이 발생할 수 있는거고요..?
몇주전에 MSA 후속 강의에 대해 문의드렸습니다. 이벤트 소싱, 보상 트랜잭션 등 후속 강의 계획이 있으신건가요? 계획이 있고, 로드맵을 알려주시면 기다릴텐데.. 아무 소식이 없이 계속 미루어지고 있습니다. 저와 같이 해당 강의를 이미 들으셨던 분, 새로 들으시는 분들 모두가 궁금해하고 듣고싶은 강의라고 생각됩니다. 후속 강의가 없다면, 해당 강의의 내용만으로는 사실 실무에서는 사용하기 어렵다고 생각됩니다. axon framework 를 별도로 공부해서 어떻게 할지 감은 오지만, 그래도 강사님이 실무에서 사용했던 방법을 공유받고 싶은 마음입니다. 꼭 좀 계획을 공유부탁드립니다. 혹시라도 후속 강의에 대한 계획이 없는거여도 말씀 좀 부탁드립니다.
안녕하세요. 강사님의 강의를 듣고, 이제는 강사님 책을 보고 있습니다. 실제 상용에서 애플리케이션을 개발할 때 궁금한 점이 있습니다. 첫번째로 프로듀서 애플리케이션 입니다. 강사님의 책에서 봤듯이, 스프링을 사용합니다. 예를 들어 저는 사용자 서비스에서 디비 트랜잭션(스프링에서 @Transactionl )을 사용하여 사용자 디비에 저장을 성공 후 프로듀서를 사용하여 레코드를 브로커에 보내야 한다고 생각합니다. 밑에 코드를 간략하게 작성하였는데, 저렇게 되면 디비는 롤백됬지만 토픽에 레코드가 저장된 상태가 될 수 있다고 생각합니다. 혹시 실무에서 커밋이 성공 후에 프로듀서 레코드를 전송하는 방법을 어떻게 하는지 간략한 코드 가 궁금합니다. 개인적인 생각은 UserService 클래스 상위 클래스로 카프카 프로듀서 처리하는 클래스를 만들어서 처리를 하는건지.. 궁금합니다.(이렇게 되면 카프카를 위한 래퍼 클래스가 항상 생기게 되는 불편함이 있는 것 같기도 하고요.. // 제가 생각하기에 잘못된 방법? // 만약 레코드 전송 후 어떤 이유로 에러가 발생하면, 디비에 저장된 데이터를 롤백되지만 // 프로듀서 레코드는 브로커 전송이 되버림 public class UserService { @Transactionl public void save(){ ... 프로듀서 레코드 전송 코드 ... 예외 발생 } } ----------------------- // 개인적인 생각 public class KafkaServce { public void save(){ userService.save(); 프로듀서 레코드 전송 코드 } } public class UserService { @Transactionl public void save(){ ... } } 두번째로 컨슈머 애플리케이션 개발 시 궁금한 점이 있습니다. 컨슈머에서 데이터 처리를하다가 어떤 이유로 에러가 발생하여 해당 레코드 처리를 계속 실패 했다고 가정합니다. 그럴 경우 이 레코드의 대한 커밋 처리를 어떻게 해야할지가 궁금합니다. 커밋을 처리하지 않으면, 다음 레코드 처리를 하지 못하는거라 생각되는데.. 어떤 방법으로 풀어내는지가 궁금합니다. 세번째로 컨슈머 애플리케이션에서 데이터베이스의 데이터를 저장해야하는 상황이다. 스프링을 사용하는 경우 기본적으로 히카리 커넥션 풀에 커넥션 10개를 생성합니다. 만약 파티션이 10개여서, 컨슈머를 10개 실행해야 한다면, 스프링 커넥션 풀을 사용하면 100개의 커넥션이 연결됩니다. 50개면 500개의 풀 계속 증가할 듯 싶네요. 이 경우 어떻게 해야할까요? 컨슈머에서 레코드들을 for 문으로 돌리기 때문에 커넥션 풀을 1개를 사용해서 개발하는게 맞는건지? 아니면 스프링에서 제공하는 히카리 커넥션 풀을 사용하지 않고, 직접 커넥션 풀을 구현하든가, 그것도 아니면 커넥션 풀을 사용하지 않고 1개의 트랜잭션당 1개의 커넥션을 생성 후 해제를 해야할까요? 강사님의 생각이 궁금합니다. 마지막으로 궁금한게 있습니다. 혹시 카프카를 활용하여 MSA에서 보상 트랜잭션(사가 패턴 - Orchestration) 처리를 할 수 있는건지 궁금합니다.
안녕하세요. 한 서버에서 producer와 consumer를 같이 구축하게 된다면, 생길 수 있는 문제가 있을까요? 혹시 현업에서 producer와 consumer를 같이 구축하는 경우가 많은지 아니면 보통 분리해서 사용하는지 궁금합니다. 만약 consumer 서버를 스케일아웃해서 3대로 운영한다면, consumer 서버에서 구독하는 모든 토픽들의 컨슈머 수가 3배가 되는게 맞을까요? 예를 들어 A, B, C 토픽을 "가consumer"서버에서 구독하고 있을때 가consumer"서버가 3대가 된다면 토픽 A <- 컨슈머 3개 / 토픽 B <- 컨슈머 3개 / 토픽 C <- 컨슈머 3개 가 되는것인지, 따로 설정해서 토픽마다 컨슈머의 수를 다르게 가져갈 수 있는 것인지 궁금합니다. 강의 잘 듣고 있습니다! 감사합니다 :)
mariadb가 아닌 oracle로 진행을 해보고 있는데 mariadb의 auto_increment 대신 oracle로 seq를 만들어 자동 증가하게 했습니다. 이렇게 적용을 하니 connector 에서 이런 오류를 주는데 오라클로 할 때는 설정이 많이 다른지 질문드립니다!!! ERROR [my-oracle-connect|task-0] WorkerSourceTask{id=my-oracle-connect-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:196) org.apache.kafka.connect.errors.ConnectException: Scale of Decimal value for incrementing column must be 0 at io.confluent.connect.jdbc.source.TimestampIncrementingCriteria.extractDecimalId(TimestampIncrementingCriteria.java:283) at io.confluent.connect.jdbc.source.TimestampIncrementingCriteria.extractOffsetIncrementedId(TimestampIncrementingCriteria.java:268) at io.confluent.connect.jdbc.source.TimestampIncrementingCriteria.extractValues(TimestampIncrementingCriteria.java:208) at io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.extractRecord(TimestampIncrementingTableQuerier.java:230) at io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:418) at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.poll(AbstractWorkerSourceTask.java:452) at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.execute(AbstractWorkerSourceTask.java:346) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:244) at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:72) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:834) [2023-12-06 19:56:27,342] INFO [my-oracle-connect|task-0] Stopping JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask:354) [2023-12-06 19:56:27,342] INFO [my-oracle-connect|task-0] Closing resources for JDBC source task (io.confluent.connect.jdbc.source.JdbcSourceTask:366) [2023-12-06 19:56:27,342] INFO [my-oracle-connect|task-0] [Producer clientId=connector-producer-my-oracle-connect-0] Closing the Kafka producer with timeoutMillis = 30000 ms. (org.apache.kafka.clients.producer.KafkaProducer:1297) [2023-12-06 19:56:27,344] INFO [my-oracle-connect|task-0] Metrics scheduler closed (org.apache.kafka.common.metrics.Metrics:693) [2023-12-06 19:56:27,344] INFO [my-oracle-connect|task-0] Closing reporter org.apache.kafka.common.metrics.JmxReporter (org.apache.kafka.common.metrics.Metrics:697) [2023-12-06 19:56:27,344] INFO [my-oracle-connect|task-0] Metrics reporters closed (org.apache.kafka.common.metrics.Metrics:703) [2023-12-06 19:56:27,344] INFO [my-oracle-connect|task-0] App info kafka.producer for connector-producer-my-oracle-connect-0 unregistered (org.apache.kafka.common.utils.AppInfoParser:83) { "name": "my-oracle-connect", "config" : { "connector.class" : "io.confluent.connect.jdbc.JdbcSourceConnector", "connection.url":"jdbc:oracle:thin:@localhost:1521:orcl", "connection.user":"test", "connection.password":"test", "mode": "incrementing", "incrementing.column.name":"ID", "table.whitelist":"USERS", "topic.prefix" : "my_oracle_", "tasks.max" : "1" } }
final KStream<String, MonitoringClass> ks0 = streamsBuilder.stream(INPUT_TOPIC, Consumed.with(STRING_SERDE, MONITORING_CLASS_SERDE)); Duration windowDuration = Duration.ofMinutes(2); TimeWindows tumblingWindow = TimeWindows.of(windowDuration.toMillis()).grace(Duration.ZERO); ks0 .groupByKey() .windowedBy(tumblingWindow) .count() .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded())) .toStream() 안녕하세요, 다음과 같이 suppress() 함수를 사용해서 윈도우가 종료되는 시점에만 특정 로직을 실행하고 싶은데 suppress() 함수를 추가한 후부터는 테스트 실행시 아예 컨슈밍이 안되는 것 처럼 보여서 도움을 얻고자 글 남겼습니다. 해당 부분도 커밋 주기 때문에 예상대로 동작하지 않는 걸까요?
@SpringBootTest class ApplyServiceTest { @Autowired private ApplyService applyService; @Autowired private CouponRepository couponRepository; @Test public void 한번만응모() { applyService.apply(1L); long count = couponRepository.count(); assertThat(count).isEqualTo(1); } @Test public void 여러번응모() throws InterruptedException { int threadCount = 1000; //ExecutorService : 병렬 작업을 간단하게 할 수 있게 도와주는 Java API ExecutorService executorService = Executors.newFixedThreadPool(32); // CountDownLatch : 다른 Thread에서 수행하는 작업을 기다리도록 도와주는 클래스 CountDownLatch latch = new CountDownLatch(threadCount); for (int i=0; i<threadCount; i++) { long userId = i; executorService.submit(() -> { try { applyService.apply(userId); } finally { latch.countDown(); } }); } latch.await(); long count = couponRepository.count(); assertThat(count).isEqualTo(100); } } @Service public class ApplyService { private final CouponRepository couponRepository; private final CouponCountRepository couponCountRepository; public ApplyService(CouponRepository couponRepository, CouponCountRepository couponCountRepository) { this.couponRepository = couponRepository; this.couponCountRepository = couponCountRepository; } public void apply(Long userId) { Long count = couponCountRepository.increment(); System.out.println("count : "+count); // long count = couponRepository.count(); if (count > 100) { return; } Coupon save = couponRepository.save(new Coupon(userId)); System.out.println("save! : "+save.getId()); } } @Repository public class CouponCountRepository { private final RedisTemplate<String, String> redisTemplate; public CouponCountRepository(RedisTemplate<String, String> redisTemplate) { this.redisTemplate = redisTemplate; } public Long increment() { try { return redisTemplate .opsForValue() .increment("coupon_count"); } catch (Exception e) { e.printStackTrace(); throw e; } } } 여러번응모 테스트 케이스 실행 CouponCountRepository.java의 increment 메서드에서 다음과 같은 에러가 발생하고 있습니다. 제일 처음에 나는 에러메시지는 다음과 같습니다. Caused by: io.lettuce.core.RedisCommandExecutionException: MISCONF Redis is configured to save RDB snapshots, but it's currently unable to persist to disk. Commands that may modify the data set are disabled, because this instance is configured to report errors during writes if RDB snapshotting fails (stop-writes-on-bgsave-error option). Please check the Redis logs for details about the RDB error. 쿠폰도 정상적으로 등록되지 않아 아에 0개로 출력되고 있습니다. 구글링을 해봤지만 도저히 원인을 찾을 수가 없어요 .. ++ 추가로, 도커에서 레디스 컨테이너와 이미지를 삭제한 후 실행해도 동일한 결과가 나타나는 것을 확인됩니다. 강의에서 나온 그대로 도커 명령어를 사용해서 레디스를 pull 하였는데 왜 컨테이너와 이미지를 삭제해도 동일한 오류가 발생하는 걸까요? 도커에서 레디스 컨테이너와 이미지를 삭제하고 ApiApplication을 실행하면 정상적으로 실행됩니다... ++ 정리 드리자면, 1. 강의를 그대로 따라했는데 RedisCommandExecutionException가 발생하면서 테스트 케이스를 실행할 수 없는 이유 2. 강의에서 나온 그대로 도커 명령어를 사용해서 레디스를 pull 하였는데 왜 컨테이너와 이미지를 삭제해도 동일한 오류가 발생하는 이유 3. 도커에서 레디스 컨테이너와 이미지를 모두 삭제하고 ApiApplication를 실행했을 때 정상적으로 실행되는 이유 답변 부탁드립니다.
안녕하세요 강사님. spring-data-redis 의존성을 추가하여 디펜전시 받고 couponcountrepository.java 를 생성하였습니다. 보통 yml 파일에 redis 관련 설정도 해주는 것으로 알고있었는데 강의에서는 따로 yml에 redis 설정을 하지 않아서요. redis 설정을 안해준다면 어플리케이션에서 어떻게 redis와 연결될 수 있는지.. 궁금합니다.
안녕하세요. producer에서 메시지가 배치 단위로 전송이 되고, retry도 해당 배치 단위로 이뤄지는 것 같은데요. 예를 들어 1~10번의 메시지가 하나의 배치를 이루고 있을 때 그 안에서 3,5번 메시지만 전송이 실패한다거나 하는 경우가 생기면 해당 메시지 배치 전체가 실패로 처리되어 retry하게 되는 걸까요? 그리고 해당 배치를 이루고 있는 내부 메시지들의 전송순서 또한 seq 기반으로 구현되어 있는 건지도 궁금합니다.
