안녕하세요 선생님실습에서 multipart-topic으로 key 없이 전송을 했는데 모조리 partition2로 가는데 어떤 설정을 살펴보면 될까요?kafka는 confluent 7.6으로 실습하고 있습니다.key값을 준 경우에는 강의 실습과 동일하게 진행되었습니다.

안녕하세요 KStreamJoinKTable 실행시 에러가 발생해서요 ㅠ 에러는 아래와 같습니다.  [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1] ERROR org.apache.kafka.streams.KafkaStreams - stream-client [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1] All stream threads have died. The instance will be in error state and should be closed. [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1] Shutdown complete Exception in thread "order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1" java.lang.UnsatisfiedLinkError: /private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib: dlopen(/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib, 0x0001): tried: '/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib' (mach-o file, but is an incompatible architecture (have 'x86_64', need 'arm64e' or 'arm64')), '/System/Volumes/Preboot/Cryptexes/OS/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib' (no such file), '/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib' (mach-o file, but is an incompatible architecture (have 'x86_64', need 'arm64e' or 'arm64')) at java.base/jdk.internal.loader.NativeLibraries.load(Native Method) at java.base/jdk.internal.loader.NativeLibraries$NativeLibraryImpl.open(NativeLibraries.java:388) at java.base/jdk.internal.loader.NativeLibraries.loadLibrary(NativeLibraries.java:232) at java.base/jdk.internal.loader.NativeLibraries.loadLibrary(NativeLibraries.java:174) at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2389) at java.base/java.lang.Runtime.load0(Runtime.java:755) at java.base/java.lang.System.load(System.java:1953) at org.rocksdb.NativeLibraryLoader.loadLibraryFromJar(NativeLibraryLoader.java:78) at org.rocksdb.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:56) at org.rocksdb.RocksDB.loadLibrary(RocksDB.java:64) at org.rocksdb.RocksDB.<clinit>(RocksDB.java:35) at org.rocksdb.DBOptions.<clinit>(DBOptions.java:21) at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:133) at org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:229) at org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48) at org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.init(ChangeLoggingKeyValueBytesStore.java:44) at org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.init(CachingKeyValueStore.java:58) at org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48) at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.lambda$init$0(MeteredKeyValueStore.java:101) at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806) at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:101) at org.apache.kafka.streams.processor.internals.AbstractTask.registerStateStores(AbstractTask.java:210) at org.apache.kafka.streams.processor.internals.StreamTask.initializeStateStores(StreamTask.java:275) at org.apache.kafka.streams.processor.internals.AssignedTasks.initializeNewTasks(AssignedTasks.java:76) at org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:397) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:779) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670) - 카프카를 공부하시면서 생긴 질문들을 남겨주세요. 상세히 작성하면 더 좋아요! - 먼저 유사한 질문이 있었는지 검색해보세요. - 서로 예의를 지키며 존중하는 문화를 만들어가요. - 잠깐! 인프런 서비스 운영 관련 문의는 1:1 문의하기를 이용해주세요.

안녕하세요. 카프카를 실무 도입전 공부를 하고 있습니다. 강의에서 3버전대는 아직 실무에 도입한 사례가 없다고 하시는데 강의가 2년전이라서 요즈음에도 아직 3버전이 크게 상용화가 안되었나요? 첫 도입 하려고 하다보니 신경쓸게 많아 질문 드립니다.감사합니다

서버 모두 기동시킨후에 했는데 runtime 오류가 납니다. 왜그러는걸까요,,? <zookeeper-server 기동><Kafka server 기동><Kafka Connect 기동> 

유레카에 등록된 게이트웨이 서비스의 인스턴스 아이디도 따로 설정을 해줘야하는지 문의드립니다. 자꾸 저렇게 뜨네요.. 포트 8000번으로 요청시에 404 에러가 뜹니다. 어떤 부분을 수정해야할까요? <사용자 조회시> <로그인시>  

userservice나 apigateway 서버 올리면 native로 저장된 파일 불러와야하는데 불러오지를 못해요.무엇을 확인해봐야 할까요,,,?config - application.yml / bootstrap.ymlUserService - bootstrap.yml

Microservcie간 통신 강의를 보다가 궁금하게 있어서 질문드립니다.user-service에서 order-service를 요청할때 인증은 필요가 없는건가요?order-service를 요청할때 따로 헤더에 토큰을 담지 않고 요청을 하고 있어서요.order-service를 IP로 지정하지 않고 유레카에 서비스 이름으로 지정해서 요청할때도 gateway를 통해서 요청이 들어 가는것인가요?

찾아보니 비슷한 질문이 있었던 거 같은데 현재 실습환경과 100% 일치하지 않아 도움을 구합니다.  실행환경은 다음과 같습니다 인텔리제이 : Build #IU-241.14494.240, built on March 28, 2024JDK : 17스프링부트 : 3.2.4 아시다 시피 최신 인텔리제이에서는 실습환경으로 제시되는 2.X 버전의 스프링부트 지원이 되지 않습니다. POM.XML로 강제로 버전을 내리거나 JDK 버전을 내리는 경우 서비스 기동이 제대로 되지 않습니다. 아마 스프링부트 3 버전에서는 JDK 특정 버전 이상을 강제하는 느낌입니다.  현재 pom.xml 설정은 다음과 같습니다. <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>3.2.4</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.example</groupId> <artifactId>user-serivce</artifactId> <version>0.0.1-SNAPSHOT</version> <name>user-serivce</name> <description>user-serivce</description> <properties> <java.version>17</java.version> <spring-cloud.version>2023.0.1</spring-cloud.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <scope>runtime</scope> <optional>true</optional> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> </dependencies> <dependencyManagement> <dependencies> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-dependencies</artifactId> <version>${spring-cloud.version}</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> <configuration> <excludes> <exclude> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </exclude> </excludes> </configuration> </plugin> </plugins> </build> </project> 명렁어 실행 시 에러는 다음과 같이 발생합니다. [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] 동일한 명렁어를 윈도우즈 cmd에서 실행하면 다음과 같습니다. [WARNING] Error injecting: org.springframework.boot.maven.RunMojojava.lang.TypeNotPresentException: Type org.springframework.boot.maven.RunMojo not present at org.eclipse.sisu.space.URLClassSpace.loadClass (URLClassSpace.java:147)(중략)Caused by: java.lang.UnsupportedClassVersionError: org/springframework/boot/maven/RunMojo has been compiled by a more recent version of the Java Runtime (class file version 61.0), this version of the Java Runtime only recognizes class file versions up to 55.0 본 문제 때문에 강의 진행이 안되고 있습니다.꼭 해결해 주셨으면 합니다.    

@Component @Slf4j public class AuthorizationHeaderFilter extends AbstractGatewayFilterFactory<AuthorizationHeaderFilter.Config> { Environment env; public AuthorizationHeaderFilter(Environment env) { super(Config.class); this.env = env; } public static class Config { // } @Override public GatewayFilter apply(Config config) { // ServerWebExchange 파라미터는 필터가 동작하는 동안 현재 요청 및 응답에 대한 정보를 제공한다. // 비동기 서버 Netty 에서는 동기 서버(ex:tomcat)와 다르게 request/response 객체를 선언할 때 Server~ 를 사용한다. GatewayFilter filter = (exchange, chain) -> { ServerHttpRequest request = exchange.getRequest(); // 요청 헤더에 "Authorization" 헤더가 포함되어 있는지 확인한다. if (!request.getHeaders().containsKey(HttpHeaders.AUTHORIZATION)) { // "Authorization" 헤더가 없는 경우, UNAUTHORIZED(401) 상태로 에러 응답을 반환. return onError(exchange, "No Authorization header", HttpStatus.UNAUTHORIZED); } // "Authorization" 헤더에서 JWT 토큰을 추출. String authorizationHeader = request.getHeaders().get(HttpHeaders.AUTHORIZATION).get(0); // "Bearer " 문자열을 삭제하고 토큰만 추출 String jwt = authorizationHeader.replace("Bearer ", ""); log.info("authorizationHeader : " + authorizationHeader); log.info("jwt : " + jwt ); // 추출한 JWT 토큰의 유효성을 확인. if (!isJwtValid(jwt)) { return onError(exchange, "JWT Token is not valid", HttpStatus.UNAUTHORIZED); } // JWT 토큰이 유효한 경우, 다음 필터로 요청을 전달. return chain.filter(exchange); }; return filter; } private boolean isJwtValid(String jwt) { // 반환값으로 사용할 boolean 변수를 초기값 true로 설정 boolean returnValue = true; // JWT의 'subject'를 저장할 변수 초기화 String subject = null; try { // 환경 변수에서 토큰의 비밀 키 가져오기 String secretKey = env.getProperty("token.secret"); log.info("secretkey : " + secretKey); // JWT 파서를 생성하고 토큰의 비밀 키 설정 JwtParser jwtParser = (JwtParser) Jwts.parser().setSigningKey(secretKey); // JWT 파서를 사용하여 토큰을 파싱하고 Claims 객체를 반환 Claims claims = jwtParser.parseClaimsJws(jwt).getBody(); // 토큰에서 'subject' 정보 추출 subject = claims.getSubject(); log.info("subject:" + subject); } catch (Exception e) { // 예외가 발생하면 JWT가 유효하지 않다고 판단하고 반환값을 false로 변경 returnValue = false; } // 'subject' 값이 비어있으면 JWT가 유효하지 않다고 판단하고 반환값을 false로 변경 if (subject == null || subject.isEmpty()) { returnValue = false; } // 최종적으로 JWT의 유효성 여부를 나타내는 반환값을 반환 return returnValue; } // Mono, Flux -> Spring WebFlux (기존의 SpringMVC 방식이 아니기때문에 Servlet 을 사용하지 않음) private Mono<Void> onError(ServerWebExchange exchange, String err, HttpStatus httpStatus) { ServerHttpResponse response = exchange.getResponse(); response.setStatusCode(httpStatus); log.error(err); return response.setComplete(); } }authorizationHeader : Bearer eyJhbGciOiJIUzI1NiJ9.eyJ1c2VybmFtZSI6ImFkbWluIiwicm9sZSI6Iltjb20uZXhhbXBsZS5qd3Rqd3QuZmlsdGVyLkN1c3RvbVVzZXJEZXRhaWxzJDFANjIxMDI1MGNdIiwiaWF0IjoxNzEyNjUxODYzLCJleHAiOjE3MTI2NTE4OTl9.oH-cw5zNVM43LIEj173W0zof5PcbhIa6CiTq1lEHRWs jwt : eyJhbGciOiJIUzI1NiJ9.eyJ1c2VybmFtZSI6ImFkbWluIiwicm9sZSI6Iltjb20uZXhhbXBsZS5qd3Rqd3QuZmlsdGVyLkN1c3RvbVVzZXJEZXRhaWxzJDFANjIxMDI1MGNdIiwiaWF0IjoxNzEyNjUxODYzLCJleHAiOjE3MTI2NTE4OTl9.oH-cw5zNVM43LIEj173W0zof5PcbhIa6CiTq1lEHRWs secretkey : user_token JWT Token is not valid이렇게 뜨는데 뭐가 문제일까요??

@Override protected void successfulAuthentication(HttpServletRequest req, HttpServletResponse res, FilterChain chain, Authentication auth) throws IOException, ServletException { String userName = ((User) auth.getPrincipal()).getUsername(); UserDto userDetails = userService.getUserDetailsByEmail(userName); byte[] secretKeyBytes = Base64.getEncoder().encode(environment.getProperty("token.secret").getBytes()); SecretKey secretKey = Keys.hmacShaKeyFor(secretKeyBytes); Instant now = Instant.now(); String token = Jwts.builder() .subject(userDetails.getUserId()) .expiration(Date.from(now.plusMillis(Long.parseLong(environment.getProperty("token.expiration_time"))))) .issuedAt(Date.from(now)) .signWith(secretKey) .compact(); res.addHeader("token", token); res.addHeader("userId", userDetails.getUserId()); } 여기서 로그인 성공 후 토큰이 발급된 뒤에 @Component@Slf4jpublic class AuthorizationHeaderFilter extends AbstractGatewayFilterFactory<AuthorizationHeaderFilter.Config> { Environment env; public AuthorizationHeaderFilter(Environment env) { super(Config.class); this.env = env; } public static class Config { // Put configuration properties here } @Override public GatewayFilter apply(Config config) { return (exchange, chain) -> { ServerHttpRequest request = exchange.getRequest(); if (!request.getHeaders().containsKey(HttpHeaders.AUTHORIZATION)) { return onError(exchange, "No authorization header", HttpStatus.UNAUTHORIZED); } HttpHeaders headers = request.getHeaders(); Set<String> keys = headers.keySet(); log.info(">>>"); keys.stream().forEach(v -> { log.info(v + "=" + request.getHeaders().get(v)); }); log.info("<<<"); String authorizationHeader = request.getHeaders().get(HttpHeaders.AUTHORIZATION).get(0); String jwt = authorizationHeader.replace("Bearer", ""); // Create a cookie object// ServerHttpResponse response = exchange.getResponse();// ResponseCookie c1 = ResponseCookie.from("my_token", "test1234").maxAge(60 60 24).build();// response.addCookie(c1); if (!isJwtValid(jwt)) { return onError(exchange, "JWT token is not valid", HttpStatus.UNAUTHORIZED); } return chain.filter(exchange); }; } private Mono<Void> onError(ServerWebExchange exchange, String err, HttpStatus httpStatus) { ServerHttpResponse response = exchange.getResponse(); response.setStatusCode(httpStatus); log.error(err); byte[] bytes = "The requested token is invalid.".getBytes(StandardCharsets.UTF_8); DataBuffer buffer = exchange.getResponse().bufferFactory().wrap(bytes); return response.writeWith(Flux.just(buffer));// return response.setComplete(); } 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.parserBuilder() .setSigningKey(signingKey) .build(); subject = jwtParser.parseClaimsJws(jwt).getBody().getSubject(); } catch (Exception ex) { returnValue = false; } if (subject == null || subject.isEmpty()) { returnValue = false; } return returnValue; }여기로 넘어 오는 게 맞나요?그리고 그 뒤에 어떻게 진행되는지 설명 부탁드립니다!!

자동 커밋 옵션을 활성화한 경우 일정 시간마다 오프셋 커밋을 실행하는 것을 이해했습니다. 결제 이벤트를 처리하는 카프카 컨슈머가 100건의 레코드를 가져와서 한번에 처리하도록 구현되어 있을 때 레코드 처리 중간에 장애가 발생하여 오프셋 커밋을 하지 못하는 경우 어떤 일이 일어나는지 궁금합니다. 50번째 레코드를 처리하다가 장애가 발생했다고 가정하면 오프셋 커밋이 되지 못했으므로 다른 컨슈머 애플리케이션이 이미 처리된 100건의 레코드를 다시 가져와서 1~50번째 레코드가 중복으로 이벤트가 처리될 것 같습니다. 이는 컨슈머 애플리케이션은 언제든지 중복으로 레코드를 소비할 수 있다는 것으로 이해가 되고, 컨슈머 애플리케이션 로직을 작성할 때는 항상 중복 레코드가 처리되지 않도록 방어해야한다는 것을 의미한다고 생각합니다. 제가 만약 방어 로직을 작성한다고 하면 레코드의 메시지마다 유일한 키값을 같이 전송하여 해당 값을 Redis에 저장하고, redis에 저장된 상태라면 메시지를 처리하지 않도록 구현할 것 같습니다. 제가 오프셋과 관련하여 이해한 내용이 정확한지, 그리고 실무에서도 Redis를 활용한 중복 처리 방지 로직이 효과적인 방법인지 궁금합니다. 또한, 이와 같은 상황을 해결하기 위해 다른 방법을 사용하시는지도 알고 싶습니다. 답변 주시면 감사하겠습니다! 

apigateway의 application.yml을 보면 get요청에는 authorizationHeaderFilter를 적용시키고 있습니다. authenticFilter는 그저 로그인 요청시 attempthAuthentication메소드와 successfulAthentication메소드를 통해 인증을 처리하고 response헤더에 토큰을 추가하는 역할입니다. 따라서 로그인 요청에만 적용하면 될 것 같은데, 깃허브의 최신코드 버전을 보면 거의 모든 요청에 authenticationFilter를 적용하고 있습니다. 왜 로그인요청이 아닌 다른 요청에다가 AthenticFilter를 적용할까요? apiGateway-service에서 GET요청만 AuthoriztionFilter를 적용하면 되지않나요??http.authorizeHttpRequests((authz) -> authz .requestMatchers(new AntPathRequestMatcher("/actuator/**")).permitAll() .requestMatchers(new AntPathRequestMatcher("/h2-console/**")).permitAll() .requestMatchers(new AntPathRequestMatcher("/users", "POST")).permitAll() .requestMatchers(new AntPathRequestMatcher("/h2-console/**")).permitAll() // .requestMatchers(new AntPathRequestMatcher("/welcome/**")).permitAll() // .requestMatchers("/**").access(this::hasIpAddress) .requestMatchers("/**").access( new WebExpressionAuthorizationManager("hasIpAddress('127.0.0.1') or hasIpAddress('172.30.1.48')")) .anyRequest().authenticated() ) 

안녕하세요 강의 잘 보고 있습니다.복제강의 수강중 궁금한점이리더 - 팔로워 간의 복제시 팔로워에서 리더의 오프셋을 확인해 차이점이 있다면 팔로워로 저장한다고 이해했는데요.이러한 복제 과정으로 인해 리더파티션에 이슈가 생길만한 케이스가 있는지 궁금합니다.기본적으로 리더 파티션이 프로듀서 - 컨슈머와의 통신을 담당하기에 팔로워 파티션에서 복제과정을 담당하게 되는 형태인것 같아 문제가 없을것 같긴한데 .. 꼭 이렇게 생각하고 넘어가면 문제가 생기더라구요 ㅎ..혹시 복제과정이 리더 파티션 성능에 영향을 주게될만한 이슈가 있을지 우려되서 질문드립니다

안녕하세요. 깃허브에서 spring boot 3.2 브랜치통해 코드를 작성했습니다. 그런데 회원가입, 로그인까지 해서 jwt토큰까지는 잘 받아오는데, 그 이후에 /welcome으로 헤더에 토큰으로 추가해서 요청을 보내도 403 forbidden에러가 뜹니다. 코드는 spring boot 3.2 브랜치와 동일합니다. 

안녕하세요! 2강에서 설명해주신 레플리케이션 설정을 통한 고가용성 외에 서버 랙, 리전 장애에도 대비할 수 있는 카프카 설정이 무엇인지 궁금하여 질문드립니다.

안녕하세요 로그인시에는 인증이 필요하지가 않는데 자꾸 인증 해주는 AuthorizationHeaderFilter 클래스에서 apply 메소드에서 header 값이 없다고 걸립니다.그래서 아래와 같이 401 오류로 로그인이 안됩니다.application.yml 파일은 문제없는거 같은데 뭘 더 확인해봐야 할까요,,,?아래는 Users application.yml 파일입니다.

docker ex -it mariadb /bin/bash이후에 mysql 명령어를 치면mysql: Command not found 에러가 나옵니다,.

[Redis 활용하며 문제 해결하기 > 문제점 해결하기 01:19]강의에서 Synchronized 를 사용할때 서버가 여러 대가 된다면 적절하지 않다고 하셨는데, 그러면 서버가 1대 일때는 Synchronized 를 사용해도 괜찮은가요??

윈도우 연산시 주의해야 할 사항에텀블링 윈도우에 한해서 윈도윙이 5초간격이라면커밋 간격도 5초로 맞춰주면굳이 최신 데이터를 어떤 로직처리에 의해서 뽑아줄 필요가 없지 않을까요?스트림 처리할때만 커밋 간격을 윈도우 크기만큼 맞추고, 나머지는 디폴트나 설정한 값으로 처리하면 될것 같은데요이렇게 하는건 어떻게 생각하시는지 궁금합니다 자바의 main 말고 카프카의 데이터를 컨슘해서 프론트엔드로 서빙하는 스프링 예제provectuslabs/kafka-ui의 이미지를 사용해서 도커컴포즈로 브로커를 3대 띄우고 운영한다던지(아무래도 쿠버네티스 환경까지 들어가면 너무 깊어지니깐)스키마 레지스트리와 KSQL 등에 대한 것추가로 2, 3, 4번 예제는 강의에 없는것 같은데 혹시 집필하신 책에 나와있을까요? 많은 내용을 배워서 갑사합니다추가로 현업에서 사용되는 다양한 처리 방법을 조금 더 심도있게 배우고 싶습니다 

안녕하세요 강사님 ! kafka: consumer: max-poll-records: 5이것 저것 찾아보다가 한번에 가져오는 레코드의 양을 줄이는 옵션이 있어서 적용해봤습니다.  그래서 처리되는 속도는 비슷하더라구요.  컨슈머가 속도를 설정해서 천천히 이벤트를 확인하고 싶은데 강사님이 말씀하신 걸 적용하려면 어떤 옵션을 줘야하나요?