왜 게이트웨이로 접근이 안될가요?hasipaddress값에 제 컴 아이피를 적고http://220.86.33.96:9295/welcome > 접근가능한데 http://220.86.33.96:8000/user-service/welcome > 접근불가떠요..이해가안됩니다 게이트웨이서버건 유저 서버건모두 제 컴퓨터고 제 서버 아이피가 220.86.33.96인데..hasIpAddress값을 127.0.0.1로 주고 접근하면 모든곳에서 접근이 됩니다..

안녕하세요고사양 pc에서는 server01, 02, 03 이 필요한 것으로 알고 있는데 혹시 server03은 어디서 받을 수 있을까요? 섹션 1~2에 업로드된 첨부파일 확인했는데 server03 이미지를 찾지 못하여 문의 드립니다!

제목이 보채는것 같이보여 죄송합니다.최대한 소심? 하게 물어보고 싶은데 단어가 안떠오르네요. 아무튼 저번 업데이트된 1편 너무 잘 봤습니다.혹시 2편 강의 내용 업데이트 날자를 대충이라도(6월 전 이렇게...) 알 수 있을까요..?

안녕하세요! 강의 잘 듣고 있습니다. 감사합니다. https://www.inflearn.com/course/lecture?courseSlug=%EC%84%A0%EC%B0%A9%EC%88%9C-%EC%9D%B4%EB%B2%A4%ED%8A%B8-%EC%8B%9C%EC%8A%A4%ED%85%9C-%EC%8B%A4%EC%8A%B5&unitId=156125&category=questionDetail&tab=community&q=1029856해당 질문과 답변을 보고 추가 질문 드리려고 합니다. 제가 kafka나 redis, 분산서버 등에 대해 이해도가 낮은 점 양해 부탁드립니다! 1.Kafka 미사용시 주문생성/회원가입요청의 타임아웃 및 10분뒤 실행에 대한 해결책으로 Kafka 를 선택한 이유는 배압조절(back pressure) 때문입니다.이렇게 말씀을 해주셨는데요,답변에서 말씀하신 예시에서 처럼 10000개 요청이 있고,카프카를 사용한다면,요청 100개가 쌓일때마다 db에 insert를 하고, 다시 요청 100개가 쌓일때까지 기다렸다가 insert 하기를 반복한다는 것으로 이해하면 될까요?2. 그게 맞다면, 강의에서 구현한 apply 메서드에서 100개의 요청이 왔는지 확인하지 않고, kafka를 사용해서 다른곳에 전달하여 처리하는 이유는 무엇인가요?예시로 apply 메서드 안에서 redis의 incr 값을 체크하면 요청이 몇개가 쌓였는지 알 수 있을테고, 데이터를 임시저장하다가 100개마다 처리할 수 있을거란 생각이 들었습니다. kafka로 다른 모듈로 전달하는 것과의 차이점이 무엇인가 궁금합니다.3.실제로 consumer에서 100개의 작업이 완료되었는지는 일반적으로 어떻게 확인하는 걸까요? db에 저장하기 전에 100개의 데이터는 어디에 임시저장을 하나요?

VirtualBox-5.0.40윈도우에서는 실행이 안된다고 에러메시지가 뜹니다. 다른분이 질문하신 게시글에서 버츄얼박스는 '흰색남자'님의 OS(Windows 10)에 최적화된 버젼을 설치해 사용 하시면 됩니다. 라고 하셨는데 그 글도 삭제된 것 같아서요 VirtualBox-5.0.40-115130-Win다른 방안을 알려주시면 감사하겠습니다.

안녕하세요. 비대칭키 암호화 강의 영상을 시청했습니다.강의를 들으면서 keystore 폴더에 keytool을 사용하여 3가지 파일을 생성했습니다.비공개키 파일: apiEncryptionKey.jks인증서 파일: trustServer.cer공개키 파일: publicKey.jks spring cloud config 프로젝트에서 다음과 같이 코드를 작성했습니다.pom.xml <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-bus-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-bootstrap</artifactId> </dependency> ConfigServiceApplication.java@SpringBootApplication @EnableConfigServer public class ConfigServiceApplication { public static void main(String[] args) { SpringApplication.run(ConfigServiceApplication.class, args); } }  bootstrap.yml encrypt: # key: abcdefghijklmnopqrstuvwxyz0123456789 key-store: location: file:///${user.home}/Desktop/keystore/apiEncryptionKey.jks password: test1234 alias: apiEncryptionKey 강의를 다 듣고 난 후 궁금한점이 생겨서 테스트를 진행했습니다.keystore 폴더에서 인증서 파일, 공개키 파일을 삭제했습니다. http://127.0.0.1:8888/encrypthttp://127.0.0.1:8888/decrypthttp://localhost:8888/ecommerce/default강의영상에 나온 위 api 호출을 다시 시도해봤습니다.마치 대칭키 암호화 방식처럼 비공개키 하나로 암호화, 복호화를 하고 있습니다. 질문1: 인증서 파일, 공개키 파일은 언제 사용하는건지 궁금합니다.질문2: spring cloud config 프로젝트에서 비대칭키 암호화를 사용하고 있는건지 궁금합니다.

"Orders Microservice에서 MariaDB 연동" 강의에서 MariaDB만 설치한 채로 mysql.server start 명령을 수행하시는데 저는 해당 명령이 없다고 나옵니다. MariaDB 바이너리 설치 경로에서 수행해도 마찬가지입니다. 설치한 버전은 강의에서 사용한 버전과 동일합니다.해당 폴더를 보면 mariadb.exe, mariadbd.exe, mysql.exe, mysqld.exe 등의 파일이 있긴 합니다.

안녕하세요, 문의 사항이 있어 글 납깁니다...당장에 이해가 안되어서요,,,,UserDetailService . loadUserByUsername()은 password활용 없이 ID만을 통해 find하게 짜도 문제 없나요,,? 어디선가 password를 따로 비교해주나요???....?어떤 컬럼인지 어떻게 알고 비교해주지,,,ㅁ 질문드립니다

안녕하세요 VM 통합 환경 구성 중입니다.현재 인텔 MAC 사용중이고 ,HOST,NAC 설정 까지 다 해주었는데 연결이 안돼서 진행을 못하고 있습니다 . ㅠ추가로 putty 접속도 안되네요.. 원인이 뭘까요??..확인 한번 부탁드립니다..++ 수정네트워크를 다음과 같이 변경후 서버 재시작하였더니이제 refused는 뜨지 않지만 time out 에러가 뜨네요 ㅠputty도 마찬가지입니다. ++ 수정 server02 는 현재 ssh 접속이 가능합니다..!정확하게 host정보를 입력한거 같은데 server01은 접속이 안되네요 ! ++ server 01 에서 바로 서비스체크 해보았습니다.클라우데라 매니저 잘 작동 중이고 ,, 리스타도 해보았는데여전히 http://server01.hadoop.com/ 치고 들어가면 refused 뜨네요 ㅜㅜ

안녕하세요제가 누락된 곳이 있는건지 테스를 시작하면터미널에서 컨슈머쪽에 숫자가 찍히지 않는데 무슨 문제일까요....?아예 실행자체가 안되는 것 같은데 혹시 터미널로 Producer테스트 코드 여쭤볼 수 있을까요..혹시 아니면 프로젝트 코드 문제가 있을 것 같아서주소 첨부드립니다..감사합니다 https://github.com/KMSKang/coupon-system  [터미널에 입력한 consumer 명령어]docker exec -it kafka kafka-console-consumer.sh --topic coupon_create --bootstrap-server localhost:9092 --key-deserializer "org.apache.kafka.common.serialization.StringDeserializer" --value-deserializer "org.apache.kafka.common.serialization.LongDeserializer"ProducerConfig values: acks = -1 batch.size = 16384 bootstrap.servers = [localhost:9092] buffer.memory = 33554432 client.dns.lookup = use_all_dns_ips client.id = producer-100 compression.type = none connections.max.idle.ms = 540000 delivery.timeout.ms = 120000 enable.idempotence = true interceptor.classes = [] key.serializer = class com.fasterxml.jackson.databind.ser.std.StringSerializer linger.ms = 0 max.block.ms = 60000 max.in.flight.requests.per.connection = 5 max.request.size = 1048576 metadata.max.age.ms = 300000 metadata.max.idle.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partitioner.adaptive.partitioning.enable = true partitioner.availability.timeout.ms = 0 partitioner.class = null partitioner.ignore.keys = false receive.buffer.bytes = 32768 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retries = 2147483647 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.connect.timeout.ms = null sasl.login.read.timeout.ms = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.login.retry.backoff.max.ms = 10000 sasl.login.retry.backoff.ms = 100 sasl.mechanism = GSSAPI sasl.oauthbearer.clock.skew.seconds = 30 sasl.oauthbearer.expected.audience = null sasl.oauthbearer.expected.issuer = null sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000 sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000 sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100 sasl.oauthbearer.jwks.endpoint.url = null sasl.oauthbearer.scope.claim.name = scope sasl.oauthbearer.sub.claim.name = sub sasl.oauthbearer.token.endpoint.url = null security.protocol = PLAINTEXT security.providers = null send.buffer.bytes = 131072 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS transaction.timeout.ms = 60000 transactional.id = null value.serializer = class org.apache.kafka.common.serialization.LongSerializer

package com.example; import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.serialization.StringDeserializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.Arrays; import java.util.Map; import java.util.Properties; public class ConsumerPartitionAssignSeek { public static final Logger logger = LoggerFactory.getLogger(ConsumerPartitionAssignSeek.class.getName()); public static void main(String[] args) { String topicName = "pizza-topic"; Properties props = new Properties(); props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.101:9092"); props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); //props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group_pizza_assign_seek_v001"); //props.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "6000"); props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props); TopicPartition topicPartition = new TopicPartition(topicName, 0); //kafkaConsumer.subscribe(List.of(topicName)); kafkaConsumer.assign(Arrays.asList(topicPartition)); kafkaConsumer.seek(topicPartition, 5L); //main thread Thread mainThread = Thread.currentThread(); //main thread 종료시 별도의 thread로 KafkaConsumer wakeup()메소드를 호출하게 함. Runtime.getRuntime().addShutdownHook(new Thread() { public void run() { logger.info(" main program starts to exit by calling wakeup"); kafkaConsumer.wakeup(); try { mainThread.join(); } catch(InterruptedException e) { e.printStackTrace();} } }); //kafkaConsumer.close(); //pollAutoCommit(kafkaConsumer); //pollCommitSync(kafkaConsumer); //pollCommitAsync(kafkaConsumer); pollNoCommit(kafkaConsumer); } private static void pollNoCommit(KafkaConsumer<String, String> kafkaConsumer) { int loopCnt = 0; try { while (true) { ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); logger.info(" ######## loopCnt: {} consumerRecords count:{}", loopCnt++, consumerRecords.count()); for (ConsumerRecord record : consumerRecords) { logger.info("record key:{}, partition:{}, record offset:{} record value:{}", record.key(), record.partition(), record.offset(), record.value()); } } }catch(WakeupException e) { logger.error("wakeup exception has been called"); }catch(Exception e) { logger.error(e.getMessage()); }finally { logger.info("finally consumer is closing"); kafkaConsumer.close(); } } private static void pollCommitAsync(KafkaConsumer<String, String> kafkaConsumer) { int loopCnt = 0; try { while (true) { ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); logger.info(" ######## loopCnt: {} consumerRecords count:{}", loopCnt++, consumerRecords.count()); for (ConsumerRecord record : consumerRecords) { logger.info("record key:{}, partition:{}, record offset:{} record value:{}", record.key(), record.partition(), record.offset(), record.value()); } kafkaConsumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) { if(exception != null) { logger.error("offsets {} is not completed, error:{}", offsets, exception.getMessage()); } } }); } }catch(WakeupException e) { logger.error("wakeup exception has been called"); }catch(Exception e) { logger.error(e.getMessage()); }finally { logger.info("##### commit sync before closing"); kafkaConsumer.commitSync(); logger.info("finally consumer is closing"); kafkaConsumer.close(); } } private static void pollCommitSync(KafkaConsumer<String, String> kafkaConsumer) { int loopCnt = 0; try { while (true) { ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); logger.info(" ######## loopCnt: {} consumerRecords count:{}", loopCnt++, consumerRecords.count()); for (ConsumerRecord record : consumerRecords) { logger.info("record key:{}, partition:{}, record offset:{} record value:{}", record.key(), record.partition(), record.offset(), record.value()); } try { if(consumerRecords.count() > 0 ) { kafkaConsumer.commitSync(); logger.info("commit sync has been called"); } } catch(CommitFailedException e) { logger.error(e.getMessage()); } } }catch(WakeupException e) { logger.error("wakeup exception has been called"); }catch(Exception e) { logger.error(e.getMessage()); }finally { logger.info("finally consumer is closing"); kafkaConsumer.close(); } } public static void pollAutoCommit(KafkaConsumer<String, String> kafkaConsumer) { int loopCnt = 0; try { while (true) { ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); logger.info(" ######## loopCnt: {} consumerRecords count:{}", loopCnt++, consumerRecords.count()); for (ConsumerRecord record : consumerRecords) { logger.info("record key:{}, partition:{}, record offset:{} record value:{}", record.key(), record.partition(), record.offset(), record.value()); } try { logger.info("main thread is sleeping {} ms during while loop", 10000); Thread.sleep(10000); }catch(InterruptedException e) { e.printStackTrace(); } } }catch(WakeupException e) { logger.error("wakeup exception has been called"); }finally { logger.info("finally consumer is closing"); kafkaConsumer.close(); } } } 해당 코드에 문제가 없는 것으로 보입니다.  java.lang.IllegalStateException: This consumer has already been closed. at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:2456) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1218) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) at com.example.ConsumerPartitionAssignSeek.pollNoCommit(ConsumerPartitionAssignSeek.java:63) at com.example.ConsumerPartitionAssignSeek.main(ConsumerPartitionAssignSeek.java:53) 해당하는 에러가 띄는데 이유를 알 수 있을까요? git 코드는 잘돌아가는 것을 확인했습니다. 차이점이 알 수가 없어서 질문드립니다,

메이븐 명령어로 프로젝트 실행시 인식을 못하는데..메이븐 버전도 나오지 않구요인텔리제이 버튼실행말고.. 명령어로 실행할때.. 인텔리제이에 내장된것으로 사용방법없나요?

안녕하세요, stream의 key.format, value.format 관련 질문있습니다.이 두 옵션이 직렬화에 대한 옵션이라고 하셨는데, select 시에는 consumer가 동작하고 insert에는 producer가 동작한다고 하셨던 것과 조금 혼동이 되어서 질문드립니다. insert 시에 stream옵션으로 직렬화가 되면 producer의 직렬화 옵션은 어떻게 되는건가요?그리고 select시 에는 topic 메시지를 consumer가 역직렬화해서 가져온 것을 다시 직렬화 하는건가요?  stream의 직렬화 시점에 대해 조금 이해가 되지 않습니다. 감사합니다. 

안녕하세요.제가 최초 학습/실습했을 때도, 겪었던 문제인데, 계속 해결을 하지 못해 해당 문제를 여쭙습니다.  [현재 문제상황]먼저 문제는, Oozie를 강제 실행했을 때, managed_smartcar_status_info 테이블은 생성되지만, 빈 테이블로, 내부에 아무 데이터가 조회되지 않는 것입니다. 제가 오타가 있을까 싶어, 테이블 삭제에 여러 날을 데이터 생성부터 재시도를 4..5번 정도 했는데, 모두 동일하게 데이터가 조회되지 않았습니다. 우선, 제가 세팅한 값과 생성되어 cat-batch-log 에 저장된 데이터의 이미지를 보여드리겠습니다.제가 작업을 실행한 것은 금일(2024-02-08)이고, 데이터 생성 시, 입력한 일자는 2022-03-22입니다. 따라서, 강의에서 말씀하신 용어를 기준으로, working_date와 biz_date는 각각, 20240208, 20220322입니다. Oozie에 task를 할당할 때는, download한 쿼리를 그대로 복사해서 붙여서 task를 정의했습니다. 이는 오타가 발생할 수 없고, 여러번 시도했었던 부분이기 때문에 해당 부분에는 에러가 없을 것으로 기대하고 있습니다. task에 대한 scheduling은 아래와 같이 구성하였습니다.Oozie를 강제 실행하였을 때, working_date에는 오늘 일자인 20240208을 입력하고, [저장] - [실행]을 했습니다. 실행 과정에서 에러는 발생하지 않았고, 로그에도 별다른 특이사항은 없었습니다. 이 후, 새로운 테이블 managed_smartcar_status_info이 생성되었음을 refresh를 통해 확인하였으나, 해당 테이블을 조회하면, 데이터가 조회되지 않는 것이 ... 일주일동안 해결하지 못하는 트러블 ...입니다.테이블을 조건절(where) 없이 조회했을 때도, 조회된 데이터가 마찬가지로 없었기 때문에 이 또한, 조건절 오타로 인해 조회되지 않는 것은 아닌 것으로 추측됩니다.   [현재 확인된 부분]hdfs 쿼리를 통해 정상적으로 파일이 생성되었음을 확인했습니다.redis에서 데이터가 저장되어있음을 확인했습니다.hue를 통한 hbase 브라우저에 DriveCarInfo 내 key값을 이용하여 데이터를 조회했을 때, 조회되는 부분을 확인하였습니다. [개인적인 의견]HiveQL/그림-6.71.hql 내 작성된 쿼리를 통해 현재 제가 겪고 있는 문제를 생각해보았는데,현재 아래 두 쿼리는 조회 값이 NULL이 아니고, 충분히 많은 데이터가 조회됩니다. 그런데,select distinct car_number from smartCar_master_over18; select distinct car_number from SmartCar_Status_Info where wrk_date = '20240208';그런데, insert 내 select 부분만 추출하여 실행하면,select t1.car_number, t1.sex, t1.age, t1.marriage, t1.region, t1.job, t1.car_capacity, t1.car_year, t1.car_model, t2.tire_fl, t2.tire_fr, t2.tire_bl, t2.tire_br, t2.light_fl, t2.light_fr, t2.light_bl, t2.light_br, t2.engine, t2.break, t2.battery, t2.reg_date, substring(t2.reg_date, 0, 8) as biz_date from SmartCar_Master_Over18 t1 join SmartCar_Status_Info t2 on t1.car_number = t2.car_number and t2.wrk_date = '20240208';조회가 되지 않는다는 것을 알게 되었습니다. 위 쿼리를 조금 더 간략히 하면,select distinct t.car_number from smartcar_master_over18 t join smartcar_status_info son t.car_number = s.car_number where s.wrk_date = '20240208'; 가 되고, 이 또한 조회된 값이 없습니다. 현재 제가 학습을 하면서 이해를 하기론, smartcar_status_info 데이터는 flume에 의해 입수된 로그 데이터이고, smartcar_master_over18는 특정 조건에 의해 redis에 저장된 일부 데이터로 알고 있는데, 이렇게 되면,smartcar_master_over18 이 smartcar_status_info 에 포함되는 관계가 필연적으로 되어야하는게 아닌가하는게 제 생각입니다. 만약 제 생각이 맞다면, 로그가 잘못 생성되었다는 뜻인데, 혹시 강사님께선 제가 생각하는 것 외에 확인해야할 부분이 있다고 생각하신다면, 조언해주시길 부탁드립니다.   항상 친절한 가르침 감사합니다.즐거운 연휴 보내시고, 새해 복 많이 받으시길 바랍니다.감사합니다. """(마음의 소리)많이 바쁘시겠지만, 가능하다면 오늘 해결 방법 및 조언을 듣고, 긴 연휴.. 프로젝트를 마무리 및 정리를 하고 싶습니다... !! ㅠㅠ """      

안녕하세요, 실시간 분석 시스템 아키텍처에 ksqlDB 사용 명분(?)을 좀 더 확실히 하고 싶어서 질문드립니다. 제가 이해한 것은 전통적 분석 시스템은 운영 DB 부하로 분석 시스템을 직접 붙일 수 없고, DW/Batch을 분석용으로 따로 두는 것으로 이해했습니다.운영계에서 DW로 데이터를 전송하는 주기가 하루 주기인 것도 마찬가지로 운영계 I/O 부하 문제인걸까요? 또한 실시간 분석 시스템의 경우 CDC를 통해 일단위 데이터 전송에서 실시간으로 전송이 가능한 것으로 보이는데 이것은 redo dump file 전송은 DB에 직접적인 부하를 주지 않기 때문에 가능한 것인가요? 마지막으로 CDC 기반으로 실시간 데이터 전송을 했을 때 타겟 DBMS가 좋은 퍼포먼스를 가져야 함은 실시간 데이터에 대한 부하를 견딜 수 있어야 하기 때문인건가요? 질문이 많네요.. 늘 좋은 강의 감사드립니다. 

안녕하세요 개발자님 에러를 해결 하지 못해 도움을 받고 싶습니다.ksqldb 마지막 강의를 마치고 실습을 하던중 mysql 테이블을topic으로 connector(avro)를 한후 ksqldb에서 table을 만드는 과정에서 타입 변환 문제가 발생 하였습니다. avro를 통해 register에 스키마를 저장 하고 사용 하고자 하였습니다.강의 해주신 .properties 설정은 하였구요.topic에 데이터 들어온느거 확인스키마 확인sqldb 테이블 생성은 되지만 검색시 밑에와 같은 에러가 발생합니다. source, sink connector 실습은 잘 되었으며, ksqldb 거치지 않고 ELK에 데이터도 잘 보내 집니다. ksqldb에서 table 생성 과정에서 PRIMARY KEY설정을 하고 생성이 됩니다. 하지만 검색을 하면 밑에와 같은 에러가 납니다.PRIMARY KEY없이 table을 생성하면 key값이 보내면 Json형태의 키로 배출됩니다. {CUSTOMER_ID=1}key 타입을 INTEGER, bigint, int 타입 해보았습니다.mysql table도 다른걸로 만들어보고 했습니다.혹시 네가 노친것이 무엇인가요?어떻게 해야 할까여? register 실행 로그를 보니 WARNING: A provider io.confluent.kafka.schemaregistry.rest.resources.SchemasResource regi stered in SERVER runtime does not implement any provider interfaces applicable in the SER VER runtime. Due to constraint configuration problems the provider io.confluent.kafka.sch emaregistry.rest.resources.SchemasResource will be ignored.Feb. 08, 2024 5:59:52 A.M. org.glassfish.jersey.internal.inject.Providers checkProviderRu ntime있습니다. 어떻게 해야 하나요?   [2024-02-08 00:47:43,983] ERROR {"type":0,"deserializationError":{"target":"key","errorMessage":"Error deserializing message from topic: mysqlavro022.oc.customers","recordB64":null,"cause":["Cannot deserialize type struct as type int32 for path: "],"topic":"mysqlavro022.oc.customers"},"recordProcessingError":null,"productionError":null,"serializationError":null,"kafkaStreamsThreadError":null} (processing.transient_OC_CUSTOMER_3798951142359913405.KsqlTopic.Source.deserializer:44)[2024-02-08 00:47:43,988] WARN stream-thread [_confluent-ksql-default_transient_transient_OC_CUSTOMER_3798951142359913405_1707320860130-b2b59a3e-3875-4eab-ad2a-185533cf65bc-StreamThread-1] task [0_0] Skipping record due to deserialization error. topic=[mysqlavro022.oc.customers] partition=[0] offset=[0] (org.apache.kafka.streams.processor.internals.RecordDeserializer:89)org.apache.kafka.common.errors.SerializationException: Error deserializing message from topic: mysqlavro022.oc.customersat io.confluent.ksql.serde.connect.KsqlConnectDeserializer.deserialize(KsqlConnectDeserializer.java:55)at io.confluent.ksql.serde.tls.ThreadLocalDeserializer.deserialize(ThreadLocalDeserializer.java:37)at io.confluent.ksql.serde.unwrapped.UnwrappedDeserializer.deserialize(UnwrappedDeserializer.java:47)at io.confluent.ksql.serde.unwrapped.UnwrappedDeserializer.deserialize(UnwrappedDeserializer.java:26)at io.confluent.ksql.serde.GenericDeserializer.deserialize(GenericDeserializer.java:59)at io.confluent.ksql.logging.processing.LoggingDeserializer.tryDeserialize(LoggingDeserializer.java:61)at io.confluent.ksql.logging.processing.LoggingDeserializer.deserialize(LoggingDeserializer.java:48)at org.apache.kafka.common.serialization.Deserializer.deserialize(Deserializer.java:60)at org.apache.kafka.streams.processor.internals.SourceNode.deserializeKey(SourceNode.java:54)at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:65)at org.apache.kafka.streams.processor.internals.RecordQueue.updateHead(RecordQueue.java:178)at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:112)at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:304)at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:968)at org.apache.kafka.streams.processor.internals.TaskManager.addRecordsToTasks(TaskManager.java:1068)at org.apache.kafka.streams.processor.internals.StreamThread.pollPhase(StreamThread.java:962)at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:751)at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:604)at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:576)Caused by: org.apache.kafka.connect.errors.DataException: Cannot deserialize type struct as type int32 for path:at io.confluent.ksql.serde.connect.ConnectDataTranslator.throwTypeMismatchException(ConnectDataTranslator.java:71)at io.confluent.ksql.serde.connect.ConnectDataTranslator.validateType(ConnectDataTranslator.java:90)at io.confluent.ksql.serde.connect.ConnectDataTranslator.validateSchema(ConnectDataTranslator.java:154)at io.confluent.ksql.serde.connect.ConnectDataTranslator.toKsqlValue(ConnectDataTranslator.java:200)at io.confluent.ksql.serde.connect.ConnectDataTranslator.toKsqlRow(ConnectDataTranslator.java:54)at io.confluent.ksql.serde.avro.AvroDataTranslator.toKsqlRow(AvroDataTranslator.java:67)at io.confluent.ksql.serde.connect.KsqlConnectDeserializer.deserialize(KsqlConnectDeserializer.java:51)... 18 more   user01@ubuntu-20:~/kafka/connector_configs/cdc_source_mysql$ register_connector cdc_source_mysql/mysql_cdc_ops_source_avro_01.jsonHTTP/1.1 201 CreatedContent-Length: 1007Content-Type: application/jsonDate: Wed, 07 Feb 2024 15:42:52 GMTLocation: http://localhost:8083/connectors/mysql_cdc_ops_source_avro_03Server: Jetty(9.4.44.v20210927){"config": {"connector.class": "io.debezium.connector.mysql.MySqlConnector","database.connectionTimezone": "Asia/Seoul","database.history.kafka.bootstrap.servers": "localhost:9092","database.history.kafka.topic": "schema-changes.mysql.oc","database.hostname": "192.168.0.26","database.include.list": "oc","database.password": "1234","database.port": "3306","database.server.id": "31002","database.server.name": "mysqlavro022","database.user": "cnt_dev","key.converter": "io.confluent.connect.avro.AvroConverter","key.converter.schema.registry.url": "http://localhost:8081","name": "mysql_cdc_ops_source_avro_03","table.include.list": "oc.customers","tasks.max": "1","time.precision.mode": "connect","transforms": "unwrap","transforms.unwrap.drop.tombstones": "false","transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState","value.converter": "io.confluent.connect.avro.AvroConverter","value.converter.schema.registry.url": "http://localhost:8081"},"name": "mysql_cdc_ops_source_avro_03","tasks": [],"type": "source"} user01@ubuntu-20:~/kafka/data/kafka-logs$ show_topic_messages avro mysqlavro022.oc.customers{"customer_id": 1}{"customer_id": 1,"email_address": "test","full_name": "test"}user01@ubuntu-20:~/kafka$ http GET http://localhost:8081/schemas{"id": 23,"schema": "{\"type\":\"record\",\"name\":\"Key\",\"namespace\":\"mysqlavro022.oc.customers\",\"fields\":[{\"name\":\"customer_id\",\"type\":\"int\"}],\"connect.name\":\"mysqlavro022.oc.customers.Key\"}","subject": "mysqlavro022.oc.customers-key","version": 1},{"id": 24,"schema": "{\"type\":\"record\",\"name\":\"Value\",\"namespace\":\"mysqlavro022.oc.customers\",\"fields\":[{\"name\":\"customer_id\",\"type\":\"int\"},{\"name\":\"email_address\",\"type\":\"string\"},{\"name\":\"full_name\",\"type\":\"string\"}],\"connect.name\":\"mysqlavro022.oc.customers.Value\"}","subject": "mysqlavro022.oc.customers-value","version": 1}, CREATE TABLE oc_customer (customer_id int PRIMARY KEY,email_address varchar,full_name varchar) WITH (KAFKA_TOPIC = 'mysqlavro022.oc.customers',KEY_FORMAT = 'AVRO',VALUE_FORMAT = 'AVRO'); ksql> describe oc_customer extended;Name : OC_CUSTOMERType : TABLETimestamp field : Not set - using <ROWTIME>Key format : AVROValue format : AVROKafka topic : mysqlavro022.oc.customers (partitions: 1, replication: 1)Statement : CREATE TABLE OC_CUSTOMER (CUSTOMER_ID INTEGER PRIMARY KEY, EMAIL_ADDRESS STRING, FULL_NAME STRING) WITH (KAFKA_TOPIC='mysqlavro022.oc.customers', KEY_FORMAT='AVRO', VALUE_FORMAT='AVRO');Field | Type------------------------------------------------CUSTOMER_ID | INTEGER (primary key)EMAIL_ADDRESS | VARCHAR(STRING)FULL_NAME | VARCHAR(STRING)------------------------------------------------Local runtime statistics------------------------  

강의에서 부하테스트를 하셨을 때 비용이 어느 정도 나왔는지 궁금합니다.부하테스트 시 비용 절약하는 방법이 궁금합니다. elb, ec2, rds 등등 설정하고, ngrinder를 통해 부하테스트를 진행한 뒤에 결과 확인 후 바로 끄면 비용을 절약할 수 있을까요?

Confluent Kafka의 커뮤니티 버전은아파치 카프카처럼 회사에서도 사용해도 문제없나요?

안녕하세요. 강의 수강 중 궁금한점이 있어 문의드립니다.   [구성]파티션 - 5개컨슈머 - 5개   [문의]1. 컨슈머 offset과 lag 관련 문의프로듀서에서 메시지를 생성해서 Kafka로 전송하면 Offset값이 2씩 증가합니다.CURRENT-OFFSET - 2증가 (7 -> 9)LOG-END-OFFSET - 2증가 (8 -> 10)메시지가 소비되고 나면 LAG은 항상 1이 유지되고 있는데 어떤 부분을 체크해 봐야할까요? (추가확인 : LAG이 항상 1이 유지되는데 kafka-console.consumer.sh로 확인해보면 메시지는 없는데 LAG이 0으로 변경됩니다.) 2. 소비된 메시지가 다시 소비되는 현상프로듀서로 메시지 생성 후 컨슈머에서 메시지를 소비하였는데 한참 시간이 지난 후 새벽시간(12시간 이후)에 이미 처리된 메시지가 컨슈머에서 다시 처리되는 현상이 발생하는데 설정값에 따라 발생할 수 있는 현상일까요?  3. 이중화 (Active-Active) 구성일 경우 컨슈머 설정이중화 구성이 되어 있는 경우 컨슈머를 @KafkaListener( concurrency = "2")로 설정하면 컨슈머는 총 4개로 운영되는 구조가 맞는지 궁금합니다.  감사합니다.

안녕하세요!강사님 강의를 듣고 msa 적용하는 프로젝트를 진행중인데, 프로젝트 controller단 전역에서 발생하는 예외를 처리하고 싶어 @@RestControllerAdvice를 활용하려고 생각중입니다.msa에서 예외를 전역으로 처리하려면 gateway에서 @RestControllerAdvice를 활용해야 될까요? 아니면 각각의 마이크로서비스마다 @RestControllerAdvice를 활용해야 될까요?둘중 어떤 설계가 더 효율적일지 답변주시면 감사하겠습니다🙏🙏