Sink connect 등록시 에러가 납니다.

Question

안녕하세요. 선생님 Confluent AVRO Install 하고 나서 sink connect 등록하고 상태를 확인 해보니 아래와 같이 에러가 발생 했습니다. ubuntu@ip-172-31-0-29:~$ curl --location --request POST ' http://localhost:8083/connectors ' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "mysql-sink-connector", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "1", "topics": "source_mysqldb.hesdp_mgr.TB_ME_TERMS_BAS", "connection.url": "jdbc:mysql://lg-mysql:3306/aurora2", "connection.user": "sink", "connection.password": "sink1234", " table.name .format": "aurora2.TB_ME_TERMS_BAS", "insert.mode": "upsert", "pk.fields": "TERMS_VER_ID", "pk.mode": "record_key", "delete.enabled": "true", "key.converter": "io.confluent.connect.avro.AvroConverter", "value.converter": "io.confluent.connect.avro.AvroConverter", "key.converter.schema.registry.url": " http://lg-schema01:8081 ", "value.converter.schema.registry.url": " http://lg-schema01:8081 " } }' {"name":"mysql-sink-connector","config":{"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector","tasks.max":"1","topics":"source_mysqldb.hesdp_mgr.TB_ME_TERMS_BAS","connection.url":"jdbc:mysql://lg-mysql:3306/aurora2","connection.user":"sink","connection.password":"sink1234"," table.name .format":"aurora2.TB_ME_TERMS_BAS","insert.mode":"upsert","pk.fields":"TERMS_VER_ID","pk.mode":"record_key","delete.enabled":"true","key.converter":"io.confluent.connect.avro.AvroConverter","value.converter":"io.confluent.connect.avro.AvroConverter","key.converter.schema.registry.url":" http://lg-schema01:8081","value.converter.schema.registry.url":"http://lg-schema01:8081","name":"mysql-sink-connector"},"tasks":[],"type":"sink "} ubuntu@ip-172-31-0-29:~$ curl -X GET http://localhost:8083/connectors/mysql-sink-connector/status | jq '.' % Total % Received % Xferd Average Speed Time Time Time Current Dload Upload Total Spent Left Speed 100 1898 100 1898 0 0 31616 0 --:--:-- --:--:-- --:--:-- 32169 { "name": "mysql-sink-connector", "connector": { "state": "RUNNING", "worker_id": "172.31.13.238:8083" }, "tasks": [ { "id": 0, "state": "FAILED", "worker_id": "172.31.13.238:8083", "trace": "java.lang.NoClassDefFoundError: com/google/common/base/Ticker at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.( CachedSchemaRegistryClient.java:175 ) at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.( CachedSchemaRegistryClient.java:158 ) at io.confluent.kafka.schemaregistry.client.SchemaRegistryClientFactory.newClient( SchemaRegistryClientFactory.java:36 ) at io.confluent.connect.avro.AvroConverter.configure( AvroConverter.java:68 ) at org.apache.kafka.connect.runtime.isolation.Plugins.newConverter( Plugins.java:297 ) at org.apache.kafka.connect.runtime.Worker.startTask( Worker.java:640 ) at org.apache.kafka.connect.runtime.Worker.startSinkTask( Worker.java:544 ) at org.apache.kafka.connect.runtime.distributed.DistributedHerder.startTask( DistributedHerder.java:1703 ) at org.apache.kafka.connect.runtime.distributed.DistributedHerder.lambda$getTaskStartingCallable$31( DistributedHerder.java:1753 ) 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 ) Caused by: java.lang.ClassNotFoundException: com.google .common.base.Ticker at java.base/java.net.URLClassLoader.findClass( URLClassLoader.java:476 ) at java.base/java.lang.ClassLoader.loadClass( ClassLoader.java:589 ) at org.apache.kafka.connect.runtime.isolation.PluginClassLoader.loadClass( PluginClassLoader.java:103 ) at java.base/java.lang.ClassLoader.loadClass( ClassLoader.java:522 ) ... 13 more " } ], "type": "sink" } 참고로 스키마 레지스토리 서버가 잘 붙나 확인 해본 결과 잘 붙습니다. ubuntu@ip-172-31-0-29:~$ curl -v lg-schema01:8081 * Trying 172.31.46.33:8081... * Connected to lg-schema01 (172.31.46.33) port 8081 (#0) > GET / HTTP/1.1 > Host: lg-schema01:8081 > User-Agent: curl/7.81.0 > Accept: / > * Mark bundle as not supporting multiuse < HTTP/1.1 200 OK < Date: Mon, 13 Mar 2023 06:14:56 GMT < Content-Type: application/vnd.schemaregistry.v1+json < Vary: Accept-Encoding, User-Agent < Content-Length: 2 < * Connection #0 to host lg-schema01 left intact {}ubuntu@ip-172-31-0-29:~$ 현재 제가 구성한 환경 정보 입니다. Apache Zookeeper 3.8.1 3개 Apache Kafka 2.13-3.3.2 3개 => Source connect : Debezium mysql connect 2.1.2 => Sink connect : Confluent JDBC Connector 10.6.3 => Avro : Confluent Avro converter 7.3 Confluent Schema registry 7.3.0 1개 mysql DB 1 개 (Source 와 Sink 는 user 로 분리해서 데이터 넣는 방식으로 테스트 중입니다) 구글링 검색 결과 Guava 를 못찾아서 그런거 같다는데, 제가 설치한 환경의 디펜던시가 문제일까요?

strato721224 · Answer

선생님, 위 에러 원인 찾다가 Kafka plug-in 설치된 곳을 가보니 /usr/local/kafka/connector-plugin/confluentinc-kafka-connect-avro-converter/lib/ 경로안에 "guava-30.1.1-jre.jar" 이 파일이 없어서 제가 추가 했습니다. 스키마에는 잘 등록 되어서 아래와 같이 나왔습니다.   ubuntu@ip-172-31-46-33:~$ http http://localhost:8081/schemas HTTP/1.1 200 OK Content-Encoding: gzip Content-Length: 195 Content-Type: application/vnd.schemaregistry.v1+json Date: Mon, 13 Mar 2023 06:50:47 GMT Vary: Accept-Encoding, User-Agent [ { "id": 2, "schema": "{"type":"record","name":"Key","namespace":"source_mysqldb.hesdp_mgr.TB_ME_TERMS_BAS_TEST","fields":[{"name":"TERMS_VER_ID","type":"string"}]," connect.name ":"source_mysqldb.hesdp_mgr.TB_ME_TERMS_BAS_TEST.Key"}", "subject": "source_mysqldb.hesdp_mgr.TB_ME_TERMS_BAS_TEST-key", "version": 1 } ]   source 쪽에 등록은 정상적으로 되었는데 sink 쪽 등록시 아래와 같이 에러가 납니다.   curl --location --request POST ' http://localhost:8083/connectors ' \ --header 'Content-Type: application/json' \ --data-raw '{ "name": "mysql-sink-connector-test", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector", "tasks.max": "1", "topics": "source_mysqldb.hesdp_mgr.TB_ME_TERMS_BAS_TEST", "connection.url": "jdbc:mysql://lg-mysql:3306/aurora2", "connection.user": "sink", "connection.password": "sink1234", " table.name .format": "aurora2.TB_ME_TERMS_BAS_TEST", "insert.mode": "upsert", "pk.fields": "TERMS_VER_ID", "pk.mode": "record_key", "delete.enabled": "true", "key.converter": "io.confluent.connect.avro.AvroConverter", "value.converter": "io.confluent.connect.avro.AvroConverter", "key.converter.schema.registry.url": " http://lg-schema01:8081 ", "value.converter.schema.registry.url": " http://lg-schema01:8081 " } }'       ubuntu@ip-172-31-0-29:/data/kafka-logs$ curl -X GET http://localhost:8083/connectors/mysql-sink-connector-test/status | jq '.' % Total % Received % Xferd Average Speed Time Time Time Current Dload Upload Total Spent Left Speed 100 2913 100 2913 0 0 176k 0 --:--:-- --:--:-- --:--:-- 284k { "name": "mysql-sink-connector-test", "connector": { "state": "RUNNING", "worker_id": "172.31.10.77:8083" }, "tasks": [ { "id": 0, "state": "FAILED", "worker_id": "172.31.10.77:8083", "trace": "org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError( RetryWithToleranceOperator.java:223 )
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute( RetryWithToleranceOperator.java:149 )
	at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord( WorkerSinkTask.java:513 )
	at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages( WorkerSinkTask.java:493 )
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll( WorkerSinkTask.java:332 )
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration( WorkerSinkTask.java:234 )
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute( WorkerSinkTask.java:203 )
	at org.apache.kafka.connect.runtime.WorkerTask.doRun( WorkerTask.java:189 )
	at org.apache.kafka.connect.runtime.WorkerTask.run ( WorkerTask.java:244 )
	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 )
Caused by: org.apache.kafka.connect.errors.DataException: Failed to deserialize data for topic source_mysqldb.hesdp_mgr.TB_ME_TERMS_BAS_TEST to Avro: 
	at io.confluent.connect.avro.AvroConverter.toConnectData( AvroConverter.java:124 )
	at org.apache.kafka.connect.storage .Converter.toConnectData( Converter.java:88 )
	at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$3( WorkerSinkTask.java:513 )
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry( RetryWithToleranceOperator.java:173 )
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError( RetryWithToleranceOperator.java:207 )
	... 13 more
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!
	at io.confluent.kafka.serializers.AbstractKafkaSchemaSerDe.getByteBuffer( AbstractKafkaSchemaSerDe.java:244 )
	at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer$DeserializationContext.<init>( AbstractKafkaAvroDeserializer.java:334 )
	at io.confluent.kafka.serializers.AbstractKafkaAvroDeserializer.deserializeWithSchemaAndVersion( AbstractKafkaAvroDeserializer.java:202 )
	at io.confluent.connect.avro.AvroConverter$Deserializer.deserialize( AvroConverter.java:172 )
	at io.confluent.connect.avro.AvroConverter.toConnectData( AvroConverter.java:107 )
	... 17 more
" } ], "type": "sink" }   질문1. 제가 Guava file 을 kafka connect avro converter 설치된 lib 에 추가 하는게 맞나요? 질문2. 바로 위에 에러 난건 Sink connect 등록시 무었을 누락 시켜서 일까요?