커넥트 에러 관련

Question

안녕하세요. 강사님 좋은 강의 잘 듣고 있습니다. 카프카 관련해서 아래와 같은 에러가 발생하여 질문 드립니다. 에러 상황은 sql insert문으로 데이터가 잘 들어가졌는데, producer를 이용하여 json구조의 데이터를 넣은 후 부터 정 상적으로 동작하지 않고, sink의 status를 보면 아래와 같은 에 러가 보입니다. 원인과 해결방법을 알려주시면 감사하겠습니다. 에러 - json구조 데이터 - {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":true,"field":"user_id"},{"type":"string","optional":true,"field":"pwd"},{"type":"string","optional":true,"field":"name"},{"type":"int64","optional":true,"name":"org.apache.kafka.connect.data.Timestamp","version":1,"field":"created_at"}],"optional":false,"name":"users"},"payload":{"id":9,"user_id":"admin12","pwd":"1111234","name":"admin","created_at":1628604236000}}

hyunlove12 · Answer

안녕하세요. 강사님 관련하여 위 에러는 producer에서 위 형식이 아닌 다른 형식으로 메시지를 보내게 되면 db insert를 하지 못하는 에러가 발생하고, 그 후에 정상적인 메시지형태(db insert할 수 있는)로 보내더라도 기존에 잘못 보냈던 메시지가 처리되지 않아 발생하는 에러 인 것으로 보입니다. sink, source를 지우고 producer에서 정상적인 형태로만 메시지를 보내보니 디비에 정상 저장되는 것을 확인하였는데, 그렇다면 위 처럼 잘못된 메시지가 온 경우 이를 무시하거나 삭제하고 다음 메시지를 이어서 처리하는 방법이 있을까요?

hyunlove12 · Answer

안녕하세요. 강사님 파일 첨부가 안돼서 아래에 텍스트로 첨부하였습니다. 불편하시겠지만 확인 부탁드리겠습니다. 감사합니다. org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:495)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:472)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:328)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:204)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
	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:748)
Caused by: org.apache.kafka.connect.errors.DataException: JsonConverter with schemas.enable requires "schema" and "payload" fields and may not contain additional fields. If you are trying to deserialize plain JSON data, set schemas.enable=false in your converter configuration.
	at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:370)
	at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:495)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
	... 13 more

Dowon Lee · Answer

안녕하세요, 이도원입니다. query를 이용해서 처리했을 때는 sink connect도 잘 작동되지만, Producer를 이용하였을 때는 오류가 발생한다는 문제는 Producer에서 전달하는 message가 DB에 저장될 수 있는 형식이 아니어서 그런것 같습니다. 위에 올려주신 에러 메시지는 확인하기에 너무 작네요.(ㅡ.ㅡ) 전체 에러 메시지를 텍스트 형식이어도 좋으니 다시 한번 첨부해 주시면 확인하는데에 도움이 될 것 같습니다. 감사합니다.