... [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] State transition from PARTITIONS_ASSIGNED to PENDING_SHUTDOWN [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] Shutting down [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] INFO org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] INFO org.apache.kafka.clients.producer.KafkaProducer - [Producer clientId=order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1-producer] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms. [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] INFO org.apache.kafka.streams.KafkaStreams - stream-client [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43] State transition from REBALANCING to ERROR [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] ERROR org.apache.kafka.streams.KafkaStreams - stream-client [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43] All stream threads have died. The instance will be in error state and should be closed. [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1] Shutdown complete Exception in thread "order-join-application-05b24bf4-65d2-4fda-83be-a754a4988a43-StreamThread-1" java.lang.UnsatisfiedLinkError: /private/var/folders/16/xqv9hsq91sn7glvzckc__r100000gn/T/librocksdbjni3612565276450787735.jnilib: dlopen(/private/var/folders/16/xqv9hsq91sn7glvzckc__r100000gn/T/librocksdbjni3612565276450787735.jnilib, 0x0001): tried: '/private/var/folders/16/xqv9hsq91sn7glvzckc__r100000gn/T/librocksdbjni3612565276450787735.jnilib' (mach-o file, but is an incompatible architecture (have 'x86_64', need 'arm64e'))샘플 코드 실행 시 해당 에러가 계속 발생하고 있습니다.원인을 알 수 있을까요?

카프카 프로듀서 소개 강의를 보다 질문드립니다.프로듀서에서 레코드를 send하면 Accumulator에서 배치로 묶는 과정을 한다고 하셨는데요.실제 Sender가 발생하는 시점에 프로듀서 애플리케이션에서 항상 1개의 배치 단위로 tcp 통신이 발생하나요?아니면 내부적으로 여러개의 배치를 한번의 tcp 통신으로 통신할까요?

안녕하세요 강의를 듣다가 궁금증이 생겼는데요. kafka는 메시지를 하나씩 처리하기 때문에 동시성 제어도 가능할 것이라고 이해했는데 그렇다면 여기서 레디스를 사용하지 않더라도 카프카만 사용해도 동시성과 관련된 데이터 정합성을 보장할 수 있는건가요?

@Configuration @EnableWebSecurity @RequiredArgsConstructor public class WebSecurity { private final UserService userService; private final BCryptPasswordEncoder bCryptPasswordEncoder; private final Environment environment; @Bean public SecurityFilterChain filterChain(HttpSecurity http) throws Exception { http.csrf(AbstractHttpConfigurer::disable) .authorizeHttpRequests(request ->{ request.requestMatchers(antMatcher("/actuator/**")).permitAll(); request.requestMatchers(antMatcher("/**")).permitAll();}) // .headers(header -> header.frameOptions( // frameOptionsConfig -> frameOptionsConfig.disable())) .apply(new MyCustomSecurity()); return http.build(); } public class MyCustomSecurity extends AbstractHttpConfigurer<MyCustomSecurity, HttpSecurity> { @Override public void configure(HttpSecurity http) throws Exception { AuthenticationManager authenticationManager = http.getSharedObject( AuthenticationManager.class); AuthenticationFilter authenticationFilter = new AuthenticationFilter(authenticationManager, userService, environment); http.addFilter(authenticationFilter); } protected void configure2(AuthenticationManagerBuilder auth) throws Exception { auth.userDetailsService(userService).passwordEncoder(bCryptPasswordEncoder); } } }해당 코드는 앞 내용까지 포함하여 설정이 적용된 버전입니다. 저같은 경우는 처음부터 mysql로 진행하여서 h2 콘솔을 사용하지 않아 frameOptions를 주석처리 하였으나, 혹여나 h2 콘솔 사용하시는 분은 주석 해제 후 사용하시면 되고, 23.08.10 기준으로 hasIpAddress는 사용 불가능합니다.

 @Configuration @EnableAspectJAutoProxy public class MetricsConfig { @Bean public TimedAspect timedAspect(MeterRegistry registry) { return new TimedAspect(registry); } }해당 config 파일을 추가해주세요.

강의에서 설명해주시기로는 쿠폰 개수를 가져오는 것부터 쿠폰 생성까지 lock을 걸어야 한다고 설명 주셨는데 이전 강의인 재고 관리 이슈와는 다르게 row가 아닌 table에 lock을 걸기 때문에 성능 이슈가 발생한다고 보면 될까요?

포스트맨으로 API_GW를 경유해서 http://127.0.0.1:8000/user-service/login 요청하면 404 에러가 발생합니다.근데 http://127.0.0.1:54656/login 로 바로 리퀘스트하면 200 상태가 반환됩니다.로그인만 저렇고 http://127.0.0.1:8000/user-service/users 요청하면 회원가입은 또 됩니다..어디 부분부터 잘못되었는지 찾아봐야할까요?필터부분도 오타없이 잘 되었는데.. - id: user-service uri: lb://USER-SERVICE predicates: - Path=/user-service/login - Method=POST filters: - RemoveRequestHeader=Cookie - RewritePath=/user-service/(?<segment>.*), /$\{segment}/ - id: user-service uri: lb://USER-SERVICE predicates: - Path=/user-service/users - Method=POST filters: - RemoveRequestHeader=Cookie - RewritePath=/user-service/(?<segment>.*), /$\{segment}/ - id: user-service uri: lb://USER-SERVICE predicates: - Path=/user-service/** - Method=GET filters: - RemoveRequestHeader=Cookie - RewritePath=/user-service/(?<segment>.*), /$\{segment}/ 

안녕하세요! 강의 재미있고 유익하게 잘 듣고 있습니다 :)확장에 대해 생각해보니, Set을 이용해서는 중복방지 이상 개수 확장에 대해서 처리는 불가능 할 거 같아서요. 혹시, 2개 이상 N개 제한에 대한 확장 방법은 어떤게 있을까요??감사합니다!

분명 현재 없는 상태이고결과가 자꾸 이상하게 나와서 sout 처리를 잠시 해보았습니다 package com.example.api.service; import com.example.api.domain.Coupon; import com.example.api.repository.CouponCountRepository; import com.example.api.repository.CouponRepository; import org.springframework.stereotype.Service; @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 applyV1(Long userId) { Long count = couponRepository.count(); if(count > 100) { return; } couponRepository.save(new Coupon(userId)); } public void applyV2(Long userId) { Long count = couponCountRepository.increment(); System.out.println(count); if(count > 100) { return; } couponRepository.save(new Coupon(userId)); } } @SpringBootTest class ApplyServiceTest { @Autowired private ApplyService applyService; @Autowired private CouponRepository couponRepository; @Test public void applyOnce() { applyService.applyV1(1L); long count = couponRepository.count(); Assertions.assertEquals(1L, count); } @Test public void 여러명응모V1() throws InterruptedException { int threadCount = 1000; ExecutorService executorService = Executors.newFixedThreadPool(32); CountDownLatch latch = new CountDownLatch(threadCount); for(int i=0; i<threadCount; i++){ long userId = i; executorService.submit(() -> { try { applyService.applyV1(userId); } catch(Exception e) { System.out.println(e); }finally { latch.countDown(); } }); } latch.await(); long count = couponRepository.count(); assertThat(count).isEqualTo(100); } @Test public void 여러명응모V2() throws InterruptedException { int threadCount = 1000; ExecutorService executorService = Executors.newFixedThreadPool(32); CountDownLatch latch = new CountDownLatch(threadCount); for(int i=0; i<threadCount; i++){ long userId = i; executorService.submit(() -> { try { applyService.applyV2(userId); } catch(Exception e) { System.out.println(e); }finally { latch.countDown(); } }); } latch.await(); long count = couponRepository.count(); org.assertj.core.api.Assertions.assertThat(count).isEqualTo(100); } } 그런데 여러명응모V2 test를 실행시에 count를 출력시다음과 같은 수가 나옵니다. 20003200112001220013200152001620017200182002020022 ??? 한번 할때마다 1000씩 쿠폰의 수가 증가중인데요;;;조회했을때는 empty라 나오는데 이렇게 되는 연유를 잘 모르갰습니다. 테스트 코드라서 rollback이 되야할거 같은데 그렇지 않는것도 잘 모르겟네요;; ㅠㅠ

* bootstrap 라이브러리를 추가하고 아래와 같이 application.yml 파일만으로 설정하면 적용이 안됩니다.dependency에 bootstrap을 의존성 추가하지 않습니다.그 후 application.yml파일에 아래와 같이 설정합니다.spring: cloud: config: name: ecommerce # yml 파일명 앞부분 config: import: optional:configserver:http://localhost:8888 # 구성정보 설정강의에서 나오는 bootstrap.yml 설정 정보를 Spring Boot 2.4버전 이후부터는 application.yml 설정 정보에서 사용할 수 있습니다. (공식홈페이지)위 방법은 공식 홈페이지 목차에서 Spring Cloud Config Client - Spring Boot Config Data Import 부분에 나와있습니다.만약 bootstrap 라이브러리를 사용한 구성설정을 하고 싶다면 공식홈페이지 목차에서 Spring Cloud Config Client - Config First Bootstrap 부분을 살펴보시면 되겠습니다.

안녕하세요, 강의 잘 봤습니다.카프카의 실제 운영관점의 질문을 드리고자 합니다. 상황에 따라 다르겠지만, 카프카 브로커의 권장 cpu 스펙이 있을까요? 카프카 브로커의 cpu사용률은 무엇에 크게 좌우될지 궁금합니다.예를들면, 토픽 및 파티션의 수에 비례한다든지, 메시지 사이즈에 비례한다든지 질문드리는 이유는 카프카를 신규 구성 예정인데요, 내부적으로 테스트 해봤을때 업무량에 크게 좌우되지는 않는 것 같았는데, 파티션 수가 많은 경우에 튀어 보이긴 합니다.  참고로 압축기능은 사용하지 않습니다.

안녕하세요. topic에 UTC 시간대로 저장되는 문제가 있습니다.해결 접근방법에 조언을 듣고 싶습니다.Sink를 적용했을때 customers, products,order_items는 문제없이 적용되었으나 orders테이블의 timestamp타입의 order_datetime컬럼에 문제가 발생하여 SMT 옵션을 추가하다가 발견한 문제입니다.결론적으로 mysql_cdc_oc_sink_orders_01.json에 "transforms": "ConvertDateTimeType", "transforms.ConvertDateTimeType.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value", "transforms.ConvertDateTimeType.target.type": "Timestamp", "transforms.ConvertDateTimeType.field": "order_datetime", "transforms.ConvertDateTimeType.format": "yyyy-MM-dd'T'HH:mm:ss'Z'", "transforms.ConvertDateTimeType.timezone": "Asia/Seoul"위 옵션을 추가하여 sink로 저장을 해결하였으나 topic에 저장되는 시간이 다르게 저장되는것을 발견했습니다.source 데이터베이스에서는 2023-06-20 13:56:40 에 저장하였으나sink 데이터베이스에서는 2023-06-20 04:56:40으로 저장되고 있었습니다.이에 topic을 확인해보니 저장되는 시간대가 2023-06-20 04:56:40으로 topic에서부터 저장되는 값이 다른 것을 알 수 있었습니다.따라서 source설정쪽이 문제일 것 같은데 "database.connectionTimeZone": "Asia/Seoul"옵션을 넣었음에도 UTC로 적용되고있어 질문드립니다. 감사합니다.    mysql_cdc_oc_source_01.json{ "name": "mysql_cdc_oc_source_01", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "192.168.56.101", "database.port": "3306", "database.user": "connect_dev", "database.password": "connect_dev", "database.server.id": "10001", "database.server.name": "mysql01", "database.include.list": "oc", "table.include.list": "oc.customers, oc.products, oc.orders, oc.order_items", "database.history.kafka.bootstrap.servers": "192.168.56.101:9092", "database.history.kafka.topic": "schema-changes.mysql.oc", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "transforms": "unwrap", "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState", "transforms.unwrap.drop.tombstones": "false", "database.connectionTimeZone": "Asia/Seoul" } }MYSQL의 TIME_ZONE은 한국시간대입니다.mysql> select @@system_time_zone; +--------------------+ | @@system_time_zone | +--------------------+ | KST | +--------------------+ 

https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.33/mysql-connector-j-8.0.33.jar제 경우 mariadb 커넥터 대신 mysql 커넥터 사용해서 해결했습니다.(... confluentXXX/share/java/kafka 폴더에 위치)정말.. 하루종일 아래 예외로 애먹었네여..Error while starting connector ...No suitable driver found for jdbc:mysql://localhost:3307/mydb mariadb 버전별 커넥터를 5개정도 해봐도 안되어서자포자기한 채로 mysql 커넥터를 사용하니 되었습니다 ㅠㅠ자세한 버전 정보입니다. (윈도우)kafka : 2.13-3.4.0kafka-connect-jdbc : 10.7.2confluent : 7.4.0mariaDB : 10.11java : 17  설정 정보입니다.confluent/etc/kafka/connect-distributed.properties89번 행 근처 plugin.path=\C:\\Work\\confluentinc-kafka-connect-jdbc-10.7.2\\lib confluent/bin/windows/kafka-run-class.bat97번 행 근처 rem classpath addition for LSB style path if exist "%BASE_DIR%\share\java\kafka\*" ( call :concat "%BASE_DIR%\share\java\kafka\*" ) (...) 122번 행 근처 rem Log4j settings IF ["%KAFKA_LOG4J_OPTS%"] EQU [""] ( if exist %~dp0../../etc/kafka/tools-log4j.properties ( set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%~dp0../../etc/kafka/tools-log4j.properties ) else ( set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%BASE_DIR%/etc/kafka/tools-log4j.properties ) ) ELSE ( rem create logs directory IF not exist "%LOG_DIR%" ( mkdir "%LOG_DIR%" ) set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%~dp0../../etc/kafka/tools-log4j.properties )  confluent/bin/windows/connect-distributed.bat28번 행 근처 rem Log4j settings IF ["%KAFKA_LOG4J_OPTS%"] EQU [""] ( set KAFKA_LOG4J_OPTS=-Dlog4j.configuration=file:%BASE_DIR%/etc/kafka/connect-log4j.properties -Dlog4j.config.dir=%BASE_DIR%/etc/kafka )POST Request{ "name":"my-source-connect", "config":{ "connector.class" : "io.confluent.connect.jdbc.JdbcSourceConnector", "connection.url":"jdbc:mysql://localhost:3307/mydb", "connection.user":"root", "connection.password":"1234", "mode":"incrementing", "incrementing.column.name":"id", "table.whitelist":"users", "topic.prefix":"my_topic_", "tasks.max":"1" } }

안녕하세요테스트를 하면서 찾아봐도 해당 부분에 대한 의문이 해결되지 않아서 문의 드립니다.예를 들어 3대의 서버에 spring event bus를 구성한 후1번 서버에서 busrefresh를 요청하면 3대 서버에서 refresh가 정상적으로 처리가 되는건 확인했는데rabbitmq에서 큐를 확인해보면 busrefresh를 요청한 1번 큐에는 메시지가 없고2번, 3번 서버에서는 큐 메시지가 들어오는데..1번에서는 busrefresh 요청을 받기 때문에 큐 메시지 없이 처리가 가능하기 때문에 큐에서 제외되는건가요?제가 확인한 내용이 맞다면 rabbmit exchange에서 topic으로 패턴 매칭되서 모든 큐에 메시지를 넣을 거 같은데..어떻게 자신의 큐만 제외시키는지 궁금합니다. 

원영님 강의를 보고 현재 스프링 프레임워크 기반의 회사 솔루션에 카프카 도입을 구현하고 있습니다. 터미널 상에서 컨슈머와 프로듀서는 잘 주고 받는데 IDE로 넘어와 적용하면 에러가 발생합니다브로커의 로그를 확인해보니 Processing automatic preferred replica leader election (kafka.controller.KafkaController)Checking need to trigger auto leader balancing (kafka.controller.KafkaController)Topics not in preferred replica for broker 0 Map() (kafka.controller.KafkaController)이런 로그들이 찍히는데 구글링해도 이렇다 할 해결방법을 못찾고 있습니다 컨슈머 프로퍼티에 부트스트랩 서버도 localhost:9092 프로듀서 프로퍼티의 부트스트랩 서버도 동일하며 서버 프로퍼티의 애드버타이즈드 리스너도 localhost:9092로 설정해뒀습니다혹시 제가 빠뜨린 부분이 있을까요? 도움 주시면 감사하겠습니다..

좋은 강의 감사드립니다.덕분에 많은 걸 배워서 마음이 든든해 집니다.그런데, 하나 궁금한 점이 있어서 여쭤보고 싶습니다.우리 개발자들은 사실 귀찮은 걸 싫어하기 때문에 기술이 나날이 발전해 가는 것이라 생각이 되는데요, 강사님이 설명주신 부분처럼 Schema나 Payload를 계속해서 수동으로 만드는 것은 개발자 입장에서는 매우 비효율적일거라 생각이 됩니다.따라서, 이미 저 부분을 해결하기 위한 많은 부분들이 고려되었을 것으로 예상이 되는데요, 혹시 저 부분을 편하게 등록할 수 있는 다른 방법이 있을까요? 키워드라도 주시면 찾아서 공부해 나가는데 큰 도움이 될 것 같습니다. 감사합니다.

안녕하세요, 철민님. 항상 좋은 강의 감사드립니다! 한 가지 질문 드리고 싶은 것이 있어 글을 작성합니다. 혹시 RocksDB에서 StateStore로 읽어온 데이터는 어디에 저장이 되어 있을까요? RocksDB는 로컬에 Key / Value 형식으로 데이터를 가지는 것으로 이해를 했는데, StateStore로 읽어오면 그 값은 메모리에 있을까요?  StateStore의 구현체가 RocksDB도 있고, In-Memory 같은 형태도 존재하고 있는데, 이렇게 구현체가 따로 있는 것으로 추정컨데 RocksDB를 이용한 StateStore는 값을 읽어와서 메모리에 따로 저장하지 않고, 필요할 때 마다 로컬 디스크에 있는 값을 바로바로 불러오는게 아닐까 싶은데... 어떤 형식으로 동작하는 것일까요...?

VM Name: Server01Failed to open/create the internal network 'HostInterfaceNetworking-VirtualBox Host-Only Ethernet Adapter' (VERR_INTNET_FLT_IF_NOT_FOUND).Failed to attach the network LUN (VERR_INTNET_FLT_IF_NOT_FOUND).Result Code:E_FAIL (0X80004005)Component:ConsoleWrapInterface:IConsole {6ac83d89-6ee7-4e33-8ae6-b257b2e81be8}  이런 오류가 뜨는데 어떻게 해결하면 좋을까요?? 구글링 해보니 host-only-ethernet을 새로만들고 다시 해보라고 하던데 그 방법도 되지 않습니다.Virtualbox 버전은 7.0.8 최신버전입니다.

안녕 하세요 궁금한 사항이 있어서 문의드립니다.Debezium CDC source connector 강의 내용중redo 로그에 commit되지 않은 데이터도 쓰여진다고 말씀해주셨는데요저보다 당연히 잘 아시겠지만 제가 알기로는 commit하면 버퍼에 쌓이고 디스크에는 반영되지 않고 그 이후에 주기적으로redo log에 쌓이는 것으로 알고 있습니다즉 commit이 되어야 redo log에 쌓이는 걸로 알고 있는데 제가 잘못알고 있는걸까요? 감사합니다.

제가 카프카에 대한 구조와 사용법을 이해하지 못해서 학습 재미가 떨어지는데요.카프카를 이렇게 써야 하는지를 맞는지가 궁금합니다. 많은 영상을 봤지만서도 사용법이 잘 이해가 안되서 질문드립니다