텍스트가 추가될 때마다 메시지를 전송하는 Producer의 Custom 객체 직렬화 구현 - 01~02강의에서 kafka서버에서 kafka-console-consumer --bootstrap-server localhost:9092 --topic order-serde-topic 로 데이터 조회 시 null이 조회되는 에러가 있습니다. 원인은 "텍스트가 추가될 때마다 메시지를 전송하는 Producer의 Custom 객체 직렬화 구현 - 01" 번 강의에서 OrderSerializer 객체에서 serialize 메소드의 serializedOrder 를 null 로 초기화 한 뒤에 값을 대입하지 않고 반환하도록 하고 지나갔네요 objectMapper.writeValueAsBytes(order); 부분을 아래와 같이 수정하면 됩니다.serializedOrder = objectMapper.writeValueAsBytes(order); 

항상 답변 감사합니다.SQL server 에서 JDBC source connector 연결 구성중에 질문있어 드립니다. 환경세팅은 강의와 유사하게 진행하였습니다.timestamp와 incrementing모드로 Config json을 구성하였고MS-SQL DB는 connect_dev 계정에 om 데이터베이스의 customers 테이블을 사용하였습니다.쿼리는 변환하여 삽입하였습니다 { "name": "mssql_jdbc_source_customers", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url": "jdbc:sqlserver://localhost:1433;databaseName=om;trustServerCertificate=true", "connection.user": "connect_dev", "connection.password": "connect_dev", "topic.prefix": "mssql_jdbc_", "table.whitelist": "om.customers", "poll.interval.ms": 10000, "mode": "timestamp+incrementing", "incrementing.column.name": "customer_id", "timestamp.column.name": "system_upd" } }config 역시 강의와 유사하나 차이점으론 trustServerCertificate=true 옵션을 줘서 ssl 인증 문제를 회피하였습니다.  이후 질문 드리고 싶은 주요 문제로는 아래의 에러메시지ERROR [mssql_jdbc_source_customers|task-0] WorkerSourceTask{id=mssql_jdbc_source_customers-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:195) org.apache.kafka.connect.errors.ConnectException: Cannot make incremental queries using timestamp columns [system_upd] on [] because all of these columns nullable. at io.confluent.connect.jdbc.source.JdbcSourceTask.validateNonNullable(JdbcSourceTask.java:546) at io.confluent.connect.jdbc.source.JdbcSourceTask.start(JdbcSourceTask.java:196) at org.apache.kafka.connect.runtime.WorkerSourceTask.initializeAndStart(WorkerSourceTask.java:225) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243) 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:829) timestamp columns [system_upd] on [] because all of these columns nullable. (null이라 문제다)가 원인이라는건데,SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, IS_NULLABLE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'customers' AND COLUMN_NAME = 'system_upd'해당 쿼리를 통해 IS_NULLABLE 컬럼이 NO로 되어있음(=null이 아님)을 확인하였습니다. 이 null 문제의 경우 다른 생각해 볼 접근방법이 뭐가 있을지 여쭤보고싶습니다.incrementing 모드에서는 토픽생성까지 성공하였습니다.감사합니다. 

안녕하세요. debezium oracle connector를 생성하고 topic 만들 시에 발생하는 오류에 관련하여 질문 있습니다.우선 오류는 [2023-05-07 20:39:26,049] INFO WorkerSourceTask{id=oracle_connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505)[2023-05-07 20:39:26,049] WARN Couldn't commit processed log positions with the source database due to a concurrent connector shutdown or restart (io.debezium.connector.common.BaseSourceTask:243)[2023-05-07 20:39:31,455] WARN Couldn't resolve server kafka:9092 from bootstrap.servers as DNS resolution failed for kafka (org.apache.kafka.clients.ClientUtils:75)[2023-05-07 20:39:31,457] INFO [Producer clientId=UK-dbhistory] Closing the Kafka producer with timeoutMillis = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer:1189)[2023-05-07 20:39:31,457] INFO WorkerSourceTask{id=oracle_connector-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:488)[2023-05-07 20:39:31,458] INFO WorkerSourceTask{id=oracle_connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505)[2023-05-07 20:39:31,459] ERROR WorkerSourceTask{id=oracle_connector-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:191)org.apache.kafka.common.KafkaException: Failed to construct kafka producer at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:441) at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:301) at io.debezium.relational.history.KafkaDatabaseHistory.start(KafkaDatabaseHistory.java:235) at io.debezium.relational.HistorizedRelationalDatabaseSchema.<init>(HistorizedRelationalDatabaseSchema.java:40) at io.debezium.connector.oracle.OracleDatabaseSchema.<init>(OracleDatabaseSchema.java:35) at io.debezium.connector.oracle.OracleConnectorTask.start(OracleConnectorTask.java:55) at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:106) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:189) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:239) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:750)Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:89) at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:48) at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:415) ... 14 more[2023-05-07 20:39:31,461] ERROR WorkerSourceTask{id=oracle_connector-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:192)[2023-05-07 20:39:31,461] INFO Stopping down connector (io.debezium.connector.common.BaseSourceTask:192)[2023-05-07 20:39:31,462] WARN Could not stop task (org.apache.kafka.connect.runtime.WorkerSourceTask:175)java.lang.NullPointerException at io.debezium.connector.oracle.OracleConnectorTask.doStop(OracleConnectorTask.java:129) at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:206) at io.debezium.connector.common.BaseSourceTask.stop(BaseSourceTask.java:176) at org.apache.kafka.connect.runtime.WorkerSourceTask.close(WorkerSourceTask.java:173) at org.apache.kafka.connect.runtime.WorkerTask.doClose(WorkerTask.java:168) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:195) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:239) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:750) 이렇게 발생하는데 topic은 생성되지 않고 connector만 생성되었습니다.kafka task를 삭제하지 못하고 있다는데, 어떻게 해야하나요? connector를 삭제하고 다시 생성해도 동일한 오류가 발생합니다. -oracle connector 설정 파일입니다-{ "name": "oracle_connector", "config": { "connector.class": "io.debezium.connector.oracle.OracleConnector", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.registry.url": "http://0.0.0.0:8081", "database.hostname": "mydb", "database.port": "port", "database.user": "user", "database.password": "userpwd", "database.dbname": "dbname", "topic.prefix": "test_", "tasks.max": "1", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.testdb", "database.server.name": "servername", "database.out.server.name": "servername" } } 

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 최신버전입니다.

강의 리뉴얼 한다고 공지 쓰셨는데 완료 일정이 어떻게될까요그때가 오면 다시한번 보려구요!

안녕하세요. 강의 끝까지 잘 들었습니다.이해한 바가 맞는지 궁금해서 문의 드립니다.카프카 커넥트데이터 파이프라인을 반복적이라고 하는 것이 1분에 한번씩 브로커로 Record를 전송하고 이를 Consumer에서 가져와서 사용하는 경우에 카프카 커넥터가 적합하다고 하신 것으로 이해하면 될까요? 반면 1회 데이터를 Producer가 전송하고 Consumer가 처리하는 경우에는 일반적인 Producer/Consumer가 적합하다는 것으로 이해했는데 맞는지 궁금합니다.Producer/Consumer or Source Connector/Sync ConnectorProducer(Source Connector)쪽이 3개의 Server, Consumer쪽이 3개의 서버로 구성된다고 할때 실제 각 Producer/Consumer는 개별 서버에서 Application을 실행시켜줘야 하는 것으로 이해가 되는데 이게 맞을까요?좋은 강의 제공해주셔서 감사합니다.

db.pool.query(`CREATE TABLE list ( id INTEGER AUTO_INCREMENT, value TEXT, PRIMARY KEY (id))`, (err, results, fields) => { console.log('err: ', err); console.log('results: ', results); });server.js 의 위 부분 실행시, 아래와 같은 오류가 출력되는데 어떤 부분을 수정해야될까요?참고로 테이블명은 강의와 다르게 list로 했고, err 확인을 위해 console.log('err: ', err); 추가했습니다.

docker-compose up 을 실행하면 아래와 같은 에러가 발생하고,nginx: [emerg] host not found in upstream "frontend:3000" in /etc/nginx/conf.d/default.conf:3docker ps 를 해봤을 때, nginx와 mysql 이름으로된 컨테이너는 있는데 frontend, backend는 없는걸로 봐서 호스트나 포트를 못찾는게 맞는 것 같긴해요.깃헙에 있는 파일들 비교했을 때는 틀린부분이 없는 것 같은데, 어떤부분을 확인해보는게 좋을지 알려주실 수 있나요 ? 구글링한 정보 를 적용해봐도 같은 에러가 나오네용..ㅠㅠ

"[참고] 크롤링과 데이터베이스 - 크롤링 코드 작성 시작 (업데이트)" 5분10초에서, G마켓 내용이 일부 변동이 발생해서 인지 sub_category가 작동되지 않습니다. sub_categories = soup.select('div.navi.group ul li a')이 부분이 작동 되지 않는 것 같은데 제가 웹크롤링이 초보라서 해결할수가 없네요 ㅠ 

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

이왕 강의를 들은 김에 자격증도 한번 생각해보려고 하는데요.강의 내용만 충실하게 들으면 자격증 시험에 도전해볼만할까요?아니면 별도로 더 준비를 해야 하는지 궁금합니다.

https://github.com/chulminkw/KafkaConnect/blob/723d598394241434f424184998176c903c0967f9/%EC%8B%A4%EC%8A%B5%EC%88%98%ED%96%89/JDBC%20Sink%20Connector%20%EC%8B%A4%EC%8A%B5.mdSource 테이블과 연계하여 Sink 테이블에 데이터 연동 테스트다른 테이블에 대해서도 Sink Connector를 생성하고 Source 테이블에 데이터 입력하여 Sink(Target) 테이블에 데이터가 동기화 되는지 확인.products_sink용 sink connector를 위해서 아래 설정을 mysql_jdbc_sink_products.json 파일에 저장.{ "name": "mysql_jdbc_sink_products", "config": { "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "1", "topics": "mysql_jdbc_products", "connection.url": "jdbc:mysql://localhost:3306/om_sink", "connection.user": "connect_dev", "connection.password": "connect_dev", "insert.mode": "upsert", "pk.mode": "record_key" "pk.fields": "product_id", "delete.enabled": "true", "table.name.format": "om_sink.products_sink", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter" } } 실습하다가 발견해서 제보드립니다.해당부분 Git에 "pk.mode": "record_key" 다음에 쉼표가 없어 register_connector 가 작동되지 않았었습니다.쉼표를 추가하니 해결되었습니다 ㅎㅎ 

kafka-console-consumer --bootstrap-server localhost:9092 --topic spooldir-test-topic --from-beginning --property print.key=true으로 명령어를 쳤을 경우 응답 값이 강사님의 경우에는 schema와 payload로 값이 나오는데 저 같은 경우에는Struct{} {"id":"1000","first_name":"Phineas","last_name":"Frede","email":"pfrederr@weather.com","gender":"Male","ip_address":"59.83.98.78","last_login":"2015-12-04T22:18:07Z","account_balance":"14095.22","country":"PK","favorite_color":"#4f2f2b"}이렇게 나오게 되는데요버전 차이 일까요? connector 설정은 아래와 같습니다{ "name": "csv_spooldir_source", "config": { "tasks.max": "3", "connector.class": "com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector", "input.path": "/home/appuser/spool_test_dir", "input.file.pattern": "^.*\\.csv", "error.path": "/home/appuser/spool_test_dir/error", "finished.path": "/home/appuser/spool_test_dir/finished", "empty.poll.wait.ms": 30000, "halt.on.error": "false", "topic": "spooldir-test-topic", "csv.first.row.as.header": "true", "schema.generation.enabled": "true" } }감사합니다.

환경 문제로 인하여 Oracle Cloud Vm 서버 3개를 별도로 띄어 실습 환경을 구성하였습니다.문제는 Hbase 에 적재가 안된다는 것입니다.일단 환경구성에서 모든 모듈들을 latest기준으로 구성을하여서 강의내용과는 다를수가 있을듯한데요.데이터 수집 적재 파이프라인 에서 kafka 까지 들어오는것은 확인하였으나 storm 토폴로지에서 kafkaspout 에서 반응이 없는것으로 봤을때 bigdata.smartcar.storm-1.0.jar 에서 바라보는 어떠한 환경설정정보가 맞지 않아 생긴문제로 생각이되어 소스를 봤으나 별다르게 찾은것은 없습니다.zookeeper cli 에서는 kafa producer 노드가 있는것은 확인했고 SmartCar-Topic 노드는 없는것으로 확인했습니다.혹시 놓친부분이 있을까요?

안녕하세요 강사님좋은 수업 감사드립니다.강의에 나온대로 코드 작성하고 travis에서도 테스트/빌드 성공했는데 aws에서 502 bad gateway 에러가 발생합니다.Dockerfile 에 EXPOSE 80 도 추가했는데 무엇이 원인인지 잘 모르겠습니다.한번 봐주시고 답변 주시면 정말 감사드리겠습니다. ㅜㅜ

고생하십니다.질문은 해당 강의 진행중에 에러가 발생하여 질문드립니다. vi ~/connector_configs/mysql_jdbc_om_source_00.json을 통한 사전 설정값{ "name": "mysql_jdbc_om_source_00", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url": "jdbc:mysql://localhost:3306/om", "connection.user": "connect_dev", "connection.password": "connect_dev", "topic.prefix": "mysql_om_", "topic.creation.default.replication.factor": 1, "topic.creation.default.partitions": 1, "catalog.pattern": "om", "table.whitelist": "om.customers", "poll.interval.ms": 10000, "mode": "incrementing", "incrementing.column.name": "customer_id" } }  명령실행값http POST http://localhost:8083/connectors @mysql_jdbc_om_source_00.json명령 결과값HTTP/1.1 201 Created Content-Length: 566 Content-Type: application/json Date: Tue, 25 Apr 2023 01:47:21 GMT Location: http://localhost:8083/connectors/mysql_jdbc_om_source_00 Server: Jetty(9.4.44.v20210927) { "config": { "catalog.pattern": "om", "connection.password": "connect_dev", "connection.url": "jdbc:mysql://localhost:3306/om", "connection.user": "connect_dev", "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "incrementing.column.name": "customer_id", "mode": "incrementing", "name": "mysql_jdbc_om_source_00", "poll.interval.ms": "10000", "table.whitelist": "om.customers", "tasks.max": "1", "topic.creation.default.partitions": "1", "topic.creation.default.replication.factor": "1", "topic.prefix": "mysql_om_" }, "name": "mysql_jdbc_om_source_00", "tasks": [], "type": "source" } 이렇게 post를 통해서 올렸을 때, CONNECT쪽에서 아래와 같은 문제가 계속 발생하고 있습니다. Either no records were produced by the task since the last offset commit, or every record has been filtered out by a transformation or dropped due to transformation or conversion errors. (org.apache.kafka.connect.runtime.WorkerSourceTask:484)connect에 에러가 발생해 일단 강의 진행을 멈춘 상태입니다. 제가 혹시 놓친 부분이 있을지 질문드립니다. 

좋은 강의 잘 보고 있습니다. 감사합니다.nginx에 대한 감이 잘 안잡혀서 그러는데,개발환경에서는 서버로 http 요청이 오면 개발서버가 그에따른 정적파일을 제공하고,운영환경에서 서버로 http 요청이오면 nginx가 그에따른 정적파일을 제공한다라고 이해하면 될까요 ?그리고 위에서 말하는 개발환경은 개발이 로컬 머신에서만 이루어지는 걸 의미하는게 맞을까요 ?

안녕하세요, 빅디님!차근차근 따라하려고 하는데.. 한 강 한강 막히고 있습니다.. 하둡과 주키퍼 명령 사용해보기 강의보며 실습중인데,명령어가 먹히지 않습니다.. 연결이 안되는 에러로 보여서 클라우데라 접속하려고 했더니 클라우데라도 접속이 안됩니다. 빅데이터 클러스터 구성 4 강의 까지는 다 따라한 상태입니다.  클라우데라 접속해서 하둡과 주키퍼 설치까지 완료하였고 일주일? 지난 후 다시 접속해서 그 이후 과정 따라하려고 하니 접속에러 & 연결에러가 뜨고 있습니다............ ㅠㅠ 막힘없이 쭉쭉 나가고 싶은데 ,, 대체 무슨문제일까요 ? 저 같은 에러를 경험하신 분이 있을까 싶어 이전 질문글을 살펴봤는데 파일럿 PC의 용량이 존재해야 한다고 하여 정리도 하고, 실행했습니다.많이 사용하긴 했지만 212GB가 남아있는 상황입니다.. 실습 쭉 따라하고 싶습니다..빅디님.....ㅠ_ㅠ!! ++로그 확인이상 없음.(참고로 모바텀을 사용하고 있어서 파일질라는 사용안하고 모바텀으로 바로 Sample.txt -> /home/bigdata 로 올렸습니다. 파일 잘 올라온 것 확인했습니다.)  감사합니다. 

안녕하세요.현재 k8s 환경에서 롤링 업데이트를 통해 팟을 배포하는 방식을 채택하여 사용중에 있습니다.메시지 컨슈밍 퍼포먼스를 줄여주기 위해 리밸런싱 시간을 줄여주는게 굉장히 중요하다는 생각이 드는데요.강의를 보다보니 Static membership과 Cooperative sticky 2가지 방식이 이에 적합한 것 같아 보입니다.실제 프로덕션 레벨에서는 어떠한 방식을 사용하는지 궁금합니다!

2가지 궁금한게 있는데요첫번째는 카프카 커넥트와 직접 코드로 구현되는 컨슈머 중 하나를 선택해야 할 때 어떤 기준으로 선택하는 것이 좋을까요?커넥트 장점을 보면 추측은 가능하지만 경험이 없으니 실무에서 어떨 때는 커넥트, 어떨 때는 컨슈머 직접 구현을 선택할 수 있는지(이건 경험 꽤 많아요) 기준이 궁금합니다 두번재는 자동으로 토픽 생성시에 시스템 종료하면 어떤게 문제가 될까요?그레이스 풀 셧다운 관련인지 어떤 문제가 생기는지 궁금합니다 감사합니다.