"not-null property references a null or transient value : com.example.orderservice.jpa.OrderEntity.orderId; nested exception is org.hibernate.PropertyValueException: not-null property references a null or transient value : com.example.orderservice.jpa.OrderEntity.orderId"이 오류는 com.example.orderservice.jpa.OrderEntity 클래스의 orderId 속성이 null 또는 임시 값으로 설정되어 있을 때 발생합니다. 즉, orderId 속성에는 null이 허용되지 않는데, null이나 임시 값으로 설정되었기 때문에 오류가 발생합니다.application.ymlserver: port: 0 spring: application: name: order-service h2: console: enabled: true settings: web-allow-others: true path: /h2-console sql: init: mode: always jpa: database-platform: org.hibernate.dialect.H2Dialect hibernate: ddl-auto: update properties: hibernate: format_sql: true show_sql: true defer-datasource-initialization: true datasource: driver-class-name: org.h2.Driver url: jdbc:h2:mem:testdb main: allow-bean-definition-overriding: true eureka: instance: instance-id: ${spring.application.name}:${spring.application.instance_id:${random.value}} client: register-with-eureka: true fetch-registry: true service-url: defaultZone: http://127.0.0.1:8761/eureka logging: level: com.example.orderservice: DEBUG OrderEntity@Data @Entity @Table(name = "orders") public class OrderEntity implements Serializable { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; @Column(nullable = false, length = 120, unique = true) private String productId; @Column(nullable = false) private Integer qty; @Column(nullable = false) private Integer unitPrice; @Column(nullable = false) private Integer totalPrice; @Column(nullable = false) private String userId; @Column(nullable = false, unique = true) private String orderId; @Column(nullable = false, updatable = false, insertable = false) @ColumnDefault(value = "CURRENT_TIMESTAMP") private Date createAt; } OrderDto@Data public class OrderDto implements Serializable { private String productId; private Integer qty; private Integer unitPrice; private Integer totalPrice; private String orderId; private String userId; } ResponseOrder@Data @JsonInclude(JsonInclude.Include.NON_NULL) public class ResponseOrder { private String productId; private Integer qty; private Integer unitPrice; private Integer totalPrice; private Date createAt; private String orderId; } OrderServicepublic interface OrderService { OrderDto createOrder(OrderDto orderDetails); OrderDto getOrderByOrderId(String orderId); Iterable<OrderEntity> getOrdersByUserId(String userId); } OrderServiceImpl@Service public class OrderServiceImpl implements OrderService{ OrderRepository orderRepository; @Autowired public OrderServiceImpl(OrderRepository orderRepository){ this.orderRepository = orderRepository; } @Override public OrderDto createOrder(OrderDto orderDto){ orderDto.setUserId(UUID.randomUUID().toString()); orderDto.setTotalPrice(orderDto.getQty() * orderDto.getUnitPrice()); ModelMapper mapper = new ModelMapper(); mapper.getConfiguration().setMatchingStrategy(MatchingStrategies.STRICT); OrderEntity orderEntity = mapper.map(orderDto, OrderEntity.class); orderRepository.save(orderEntity); OrderDto returnValue = mapper.map(orderEntity, OrderDto.class); return returnValue; } @Override public OrderDto getOrderByOrderId(String orderId) { OrderEntity orderEntity = orderRepository.findByOrderId(orderId); OrderDto orderDto = new ModelMapper().map(orderEntity, OrderDto.class); return orderDto; } @Override public Iterable<OrderEntity> getOrdersByUserId(String userId) { return orderRepository.findByUserId(userId); } }OrderController@RestController @RequestMapping("/order-service") public class OrderController { Environment env; OrderService orderService; @Autowired public OrderController(Environment env, OrderService orderService) { this.env = env; this.orderService = orderService; } @GetMapping("/health_check") public String status(){ return String.format("It's Working in Order Service on PORT %s", env.getProperty("local.server.port")); } // http://127.0.0.1:0/order-service/{user_id}/orders/ @PostMapping("/{userId}/orders") public ResponseEntity<ResponseOrder> createOrder(@PathVariable("userId") String userId, @RequestBody RequestOrder orderDetails){ ModelMapper mapper = new ModelMapper(); mapper.getConfiguration().setMatchingStrategy(MatchingStrategies.STRICT); OrderDto orderDto = mapper.map(orderDetails, OrderDto.class); orderDto.setUserId(userId); OrderDto createOrder = orderService.createOrder(orderDto); ResponseOrder responseOrder = mapper.map(createOrder, ResponseOrder.class); return ResponseEntity.status(HttpStatus.CREATED).body(responseOrder); } @GetMapping("/{userId}/orders") public ResponseEntity<List<ResponseOrder>> getOrder(@PathVariable("userId") String userId){ Iterable<OrderEntity> orderList = orderService.getOrdersByUserId(userId); List<ResponseOrder> result = new ArrayList<>(); orderList.forEach(v -> { result.add(new ModelMapper().map(v, ResponseOrder.class)); }); return ResponseEntity.status(HttpStatus.OK).body(result); } }  다 똑같이 쳤는데 뭐가 문제일까요?

안녕하세요 멱등성 프로듀서를 사용할꺼면기본 값이 retry가 INTEGER MAX, acks=all로 설정이 되는데꼭 이 기본 값을 같이 사용해야지만 멱등성 프로듀서를 사용할 수 있는건가요??강의 마지막에 Consumer에서 단 한 번만 데이터를 처리할 수 있게 만들 수 있다고 하셨는데 이건 어떻게 하는 건지 알 수 있을까요?

분명 현재 없는 상태이고결과가 자꾸 이상하게 나와서 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이 되야할거 같은데 그렇지 않는것도 잘 모르겟네요;; ㅠㅠ

파티션 개수를 늘리면프로듀서의 처리량이 늘어나는 게 맞을까요?이게 맞다면 왜 파티션 개수를 늘리면프로듀서의 처리량이 늘어나는 지 알 수 있을까요?파티션이 2개라고 했을 때 프로듀서의 데이터를 넣는 게파티션 A 넣고 파티션 B 넣고 파티션 A 넣고 이렇게 되면파티션 하나 일 때랑 별 차이가 없을 것 같아서 질문 드립니다.

안녕하세요강의를 듣다보니CommitAsync는 commitSync에 비해서 데이터 처리가 일시적으로 중단되지 않아서 더 좋아보이는데CommitAsync의 단점은 없을까요?운영환경에서도 보통 CommitAsync를 사용하시나요?

kafka Producer를 사용할 때 Key 값을 사용하거나 커스텀 파티셔너를 사용하면 특정 파티션으로만 데이터를 넣을 수 있는데 특정 파티션으로 데이터를 넣는 가장 큰 이유가데이터의 순서를 보장하기 위해서 인가요??아니면 다른 이유도 있을까요?? 강의를 들어보면서 순서가 중요하지 않다면Producer key나 커스텀 파티셔너를 안 사용해도 될 것 같아서 질문드립니다.

안녕하세요LAG가 있을 때컨슈머를 추가하는 것도 해결 방법 중 하나라고 하셨는데정확히 어떻게 추가를 하는건지 알 수 있을까요? 예를 들어Kafka 토픽 : test-topic기존 컨슈머 그룹 이름 : test-consumer-group기존 컨슈머 : 127.0.0.1 서버에서 a.jar 파일을 사용하는 컨슈머이고 test-consumer-group에 속해 있음 여기서 컨슈머를 추가하면127.0.0.2 신규 서버를 하나 만들고 기존과 똑같이 a.jar 파일을 사용하면서 기존 test-consumer-group에 추가하고 test-topic을 컨슘하게 만들기만 하면 되는건가요?? 

* 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 부분을 살펴보시면 되겠습니다.

좋은 강의에 항상 신경써주셔서 감사합니다미처 수강하지 못한 부분이 있어서 수강을 하고자 합니다너무 좋은 강의에 감사드리며 요청드립니다

안녕하세요세그먼트가 00000.log, 00010.log, 00020.log가 있다고 하고 Consumer가 30번 offset까지 읽었을 때offset 10 ~ 19번까지를 다시 읽고 싶으면00010.log 파일을 사용하면 되는걸까요??

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

안녕하세요. 강의를 들으면서 코드를 실행하다가 막힌 부분이 있어 질문드립니다.특정 도구에서 사용하면 서버가 실행이 되지 않습니다.인텔리제이 터미널에서 실행한 경우 에러가 발생합니다.PS C:\workplace\user-service> mvn spring-boot:run -Dspring-boot.run.jvmArguments='-Dserver.port=9003' -e[INFO] Error stacktraces are turned on.[INFO] Scanning for projects...[INFO] [INFO] ----------------------< com.example:user-service >----------------------[INFO] Building user-service 0.0.1-SNAPSHOT[INFO] from pom.xml[INFO] --------------------------------[ jar ]---------------------------------[INFO] ------------------------------------------------------------------------[INFO] BUILD FAILURE[INFO] ------------------------------------------------------------------------[INFO] Total time: 0.326 s[INFO] Finished at: 2023-06-27T21:49:08+09:00[INFO] ------------------------------------------------------------------------[ERROR] Unknown lifecycle phase ".run.jvmArguments=-Dserver.port=9003". You must specify a valid lifecycle phase or a goal in the format <plugin-prefix>:<goal> or <plugin-group-id>:<plugin-artifact-id>[:<plugin-version>]:<goal>. Available lifecycle phases are: pre-clean, clean, post-clean, validate, initialize, generate-sources, process-sources, generate-resources, process-resources, compile, process-classes, generate-test-sources, process-test-sources, generate-test-resources, process-test-resources, test-compile, process-test-classes, test, prepare-package, package, pre-integration-test, integration-test, post-integration-test, verify, install, deploy, pre-site, site, post-site, site-deploy. -> [Help 1]org.apache.maven.lifecycle.LifecyclePhaseNotFoundException: Unknown lifecycle phase ".run.jvmArguments=-Dserver.port=9003". You must specify a valid lifecycle phase or a goal in the format <plugin-prefix>:<goal> or <plugin-group-id>:<plugin-artifact-id>[:<plugin-version>]:<goal>. Available lifecycle phases are: pre-clean, clean, post-clean, validate, initialize, generate-sources, process-sources, generate-resources, process-resources, compile, process-classes, generate-test-sources, process-test-sources, generate-test-resources, process-test-resources, test-compile, process-test-classes, test, prepare-package, package, pre-integration-test, integration-test, post-integration-test, verify, install, deploy, pre-site, site, post-site, site-deploy. at org.apache.maven.lifecycle.internal.DefaultLifecycleExecutionPlanCalculator.calculateLifecycleMappings (DefaultLifecycleExecutionPlanCalculator.java:225) at org.apache.maven.lifecycle.internal.DefaultLifecycleExecutionPlanCalculator.calculateMojoExecutions (DefaultLifecycleExecutionPlanCalculator.java:197) at org.apache.maven.lifecycle.internal.DefaultLifecycleExecutionPlanCalculator.calculateExecutionPlan (DefaultLifecycleExecutionPlanCalculator.java:119) at org.apache.maven.lifecycle.internal.DefaultLifecycleExecutionPlanCalculator.calculateExecutionPlan (DefaultLifecycleExecutionPlanCalculator.java:135) at org.apache.maven.lifecycle.internal.builder.BuilderCommon.resolveBuildPlan (BuilderCommon.java:93) at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject (LifecycleModuleBuilder.java:100) at org.apache.maven.lifecycle.internal.LifecycleModuleBuilder.buildProject (LifecycleModuleBuilder.java:73) at org.apache.maven.lifecycle.internal.builder.singlethreaded.SingleThreadedBuilder.build (SingleThreadedBuilder.java:53) at org.apache.maven.lifecycle.internal.LifecycleStarter.execute (LifecycleStarter.java:118) at org.apache.maven.DefaultMaven.doExecute (DefaultMaven.java:261) at org.apache.maven.DefaultMaven.doExecute (DefaultMaven.java:173) at org.apache.maven.DefaultMaven.execute (DefaultMaven.java:101) at org.apache.maven.cli.MavenCli.execute (MavenCli.java:906) at org.apache.maven.cli.MavenCli.doMain (MavenCli.java:283) at org.apache.maven.cli.MavenCli.main (MavenCli.java:206) at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0 (Native Method) at jdk.internal.reflect.NativeMethodAccessorImpl.invoke (NativeMethodAccessorImpl.java:62) at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke (DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke (Method.java:566) at org.codehaus.plexus.classworlds.launcher.Launcher.launchEnhanced (Launcher.java:283) at org.codehaus.plexus.classworlds.launcher.Launcher.launch (Launcher.java:226) at org.codehaus.plexus.classworlds.launcher.Launcher.mainWithExitCode (Launcher.java:407) at org.codehaus.plexus.classworlds.launcher.Launcher.main (Launcher.java:348)[ERROR][ERROR] Re-run Maven using the -X switch to enable full debug logging.[ERROR][ERROR] For more information about the errors and possible solutions, please read the following articles:[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/LifecyclePhaseNotFoundException PowerShell에서 실행한 경우 에러가 발생합니다.PS C:\workplace\user-service> java -jar -Dserver.port=9004 .\target\user-service-0.0.1-SNAPSHOT.jarError: Unable to access jarfile .port=9004※ PowerShell에서 실행한 명령어를 명령 프롬프트에서 실행하면 정상작동됩니다.감사합니다.

/14523769<;B@C8:ED?A=>FKHLOGMNQRISJPTWZ^XUb_`]a\[YVd 소스코드 내에서 완전히 동일하게 따라했는데 왜 이렇게 나타나는지 혹시 아시나요..

클러스터 3노드(VM 3개 띄울 예정)로 구성하고자 합니다. confluent kafka로 3노드의 클러스터 구성이 가능한가요??? 확인 부탁드립니다. 감사합니다...

 안녕하세요 .. 위 부분에서 apply메소드는 raceCondition으로 인하여 정합성이 어긋나야하는데왜 자꾸 assertEquals가 100으로 되며 성공하는걸까요.. apply에 Thead.sleep(10)을 넣어도 ExcutorService thread 개수를 1로 하여고아니면 requestCount 개수를 1000으로 늘려도 무조건 성공이 됩니다.. 

원영님 안녕하세요. 강의 항상 잘듣고있습니다.다름아니라, CDC커넥터에서 발생된 메시지 처리시 '메세지 역전현상'을 해결하기 위해 테이블의 PK값을 기준으로 특정 파티션으로 데이터가 보내질수 있도록 cusomPartitioner을 구현하고 commitSync로 메시지 순서를 보장하려고 하는데요.ex) 1번 pk -> 1번 파티션 -> 1번 consumer.2번 pk -> 2번 파티션 -> 2번 consumer.이럴 경우에 특정파티션으로 데이터가 쏠리면 스케일아웃을 한 의미가 없어지더라구요. 실시간성이 요구되는 비지니스라 일정 이상의 TPS가 나와야되거든요. 문제 접근방식이 잘못됬는지... 메시지역전문제와 TPS 성능을 위해선 컨슈머에서 다른 도구들을 이용해서 해결해야 하는건지... 확신이 안서더라구요.. 혹시 비슷한 경험을 통해서 위와같은 문제를 해결해보셨다면 조언부탁드리겠습니다. 감사합니다! 

다른 예제들을 보면spring: redis: host: localhost port: 6379이런식으로 셋팅해서 사용하던데왜 여기에는 이런게 필요가 없는 건가요?처음 접해봐서 모르겠씁니다ㅠㅠ..

안녕하세요! "5. 빅데이터 클러스터 구성 4 - putty로 서버환경 접속"을 수강중에 있습니다.putty server01 login : root는 입력되는데,password가 쳐지지 않고 조금 있으면 접속이 끊긴다는 오류 메세지가 출력됩니다."Remote side unexpectedly closed network connection"어떻게 조치해야 하나요?

로그인 시도 시 401 오류를 반환하고 있습니다.user-service의 AuthenticationFilter 에서 인증이 실패가 된지 알아 디버깅을 해보니 인증이 성공해서 도대체 어디가 문제인지 알 수가 없습니다.그래서 유저 서비스에 직접 접근을 할 시 똑같은 401 오류를 반환하고 있습니다.그리고 유저 서비스에 직접 접근을 하거나 게이트웨이로 접근해도 성공에 관한 메소드는 작동하지 않습니다.< 스프링 부트, 시큐리티는 현재 최신 버전입니다. > AuthenticationFilter 클래스@RequiredArgsConstructor @Slf4j public class AuthenticationFilter extends UsernamePasswordAuthenticationFilter { private final UserService userService; @Value("${token.expiration_time}") private Long EXPIRATION_TIME; @Value("${token.secret}") private String secretKey; @Override public Authentication attemptAuthentication(HttpServletRequest request, HttpServletResponse response) throws AuthenticationException { log.info("attemptAuthentication"); try{ RequestLoginVo cred=new ObjectMapper().readValue(request.getInputStream(), RequestLoginVo.class); log.info(cred.getEmail()+" "+cred.getPassword()); return this.getAuthenticationManager().authenticate(new UsernamePasswordAuthenticationToken( cred.getEmail(), cred.getPassword(), new ArrayList<>() )); } catch(IOException e){ throw new RuntimeException(e); } } @Override protected void successfulAuthentication(HttpServletRequest request, HttpServletResponse response, FilterChain chain, Authentication authResult) throws IOException, ServletException { String userName = ((User)authResult.getPrincipal()).getUsername(); UserDto userDto = userService.getUserDetailByEmail(userName); String token = Jwts.builder() .setSubject(userDto.getUserId()) .setExpiration(new Date(System.currentTimeMillis() + EXPIRATION_TIME)) .signWith(SignatureAlgorithm.HS512, secretKey) .compact(); response.addHeader("token",token); response.addHeader("userId",userDto.getUserId()); } }loadUserByUsername 메소드@Override public UserDetails loadUserByUsername(String username) throws UsernameNotFoundException { log.info("loadUserByUsername "+username); Optional<UserEntity> userEntity = userRepository.findByEmail(username); log.info(userEntity.get().getEmail()+" "+userEntity.get().getName()+" "+userEntity.get().getEncryptedPwd()); if(userEntity.isEmpty()){ throw new UsernameNotFoundException("해당 유저는 존재하지 않습니다."); } return new User(userEntity.get().getEmail(), userEntity.get().getEncryptedPwd(), true, true, true, true, new ArrayList<>()); }WebSecurity 클래스@Configuration @EnableWebSecurity @RequiredArgsConstructor public class WebSecurity{ private final BCryptPasswordEncoder bCryptPasswordEncoder; private final UserService userService; @Bean public SecurityFilterChain filterChain(HttpSecurity http) throws Exception { http .csrf(csrf -> csrf.disable()) .headers(headers -> headers .frameOptions((frameOptions) -> frameOptions.disable())) .authorizeRequests(authorize -> authorize .requestMatchers(PathRequest.toH2Console()).permitAll() .requestMatchers("/**").hasIpAddress("127.0.0.1") ) .addFilter(getAuthenticationFilter()); return http.build(); } @Bean public AuthenticationManager authenticationManager() { DaoAuthenticationProvider provider = new DaoAuthenticationProvider(); provider.setUserDetailsService(userService); provider.setPasswordEncoder(bCryptPasswordEncoder); return new ProviderManager(provider); } private AuthenticationFilter getAuthenticationFilter() { AuthenticationFilter authenticationFilter = new AuthenticationFilter(userService); authenticationFilter.setAuthenticationManager(authenticationManager()); return authenticationFilter; } }ApiGateWay - Application.yml 유저 서비스 부분 routes: - 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} - AuthorizationHeaderFilter

안녕하세요. 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 | +--------------------+