[Bug] [SeaTunnel] Kafka to MySQL

[Hu-WF]

Search before asking

• [X] I had searched in the issues and found no similar issues.

What happened

kafka2mysql task run failed：

SeaTunnel Version

2.3.3

SeaTunnel Config

pass

Running Command

create kafka2mysql task based on seatunnel-web

Error Exception

cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit
java.lang.ClassCastException: cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit
![image](https://github.com/apache/seatunnel/assets/30825060/6d3dd4da-f3e3-46f1-8e3b-208e27764453)

Zeta or Flink or Spark Version

Zeta

Java or Scala Version

Java

Screenshots

No response

Are you willing to submit PR?

• [ ] Yes I am willing to submit a PR!

Code of Conduct

• [X] I agree to follow this project's Code of Conduct

[Hu-WF]

2024-01-09 20:05:15,572 INFO org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask - starting seatunnel source task, index 0 2024-01-09 20:05:15,575 INFO org.apache.kafka.common.security.authenticator.AbstractLogin - Successfully logged in. 2024-01-09 20:05:15,578 WARN org.apache.kafka.clients.admin.AdminClientConfig - The configuration 'sasl.jaas.config' was supplied but isn't a known config. 2024-01-09 20:05:15,578 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 2.7.1 2024-01-09 20:05:15,578 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: 61dbce85d0d41457 2024-01-09 20:05:15,578 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1704801915578 2024-01-09 20:05:15,673 INFO org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask - received reader register, readerID: TaskLocation{taskGroupLocation=TaskGroupLocation{jobId=797073430418554881, pipelineId=1, taskGroupId=30000}, taskID=40000, index=0} 2024-01-09 20:05:15,771 INFO org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask - received enough reader, starting enumerator... 2024-01-09 20:05:15,771 INFO org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplitEnumerator - Discovered topics: [dgwire_da] 2024-01-09 20:05:15,788 ERROR org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation - [localhost]:5801 [seatunnel] [5.1] cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit java.lang.ClassCastException: cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2301) ~[?:1.8.0_292] at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1431) ~[?:1.8.0_292] at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2410) ~[?:1.8.0_292] at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2328) ~[?:1.8.0_292] at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2186) ~[?:1.8.0_292] at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1666) ~[?:1.8.0_292] at java.io.ObjectInputStream.readObject(ObjectInputStream.java:502) ~[?:1.8.0_292] at java.io.ObjectInputStream.readObject(ObjectInputStream.java:460) ~[?:1.8.0_292] at org.apache.seatunnel.common.utils.SerializationUtils.deserialize(SerializationUtils.java:74) ~[connector-kudu-2.3.3.jar:2.3.3] at org.apache.seatunnel.api.serialization.DefaultSerializer.deserialize(DefaultSerializer.java:41) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.api.serialization.DefaultSerializer.deserialize(DefaultSerializer.java:25) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation.lambda$run$0(AssignSplitOperation.java:67) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48) ~[connector-kudu-2.3.3.jar:2.3.3] at org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation.run(AssignSplitOperation.java:54) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.Operation.call(Operation.java:189) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.call(OperationRunnerImpl.java:273) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:248) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:213) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:175) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:139) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102) ~[seatunnel-starter.jar:2.3.3] 2024-01-09 20:05:15,789 WARN org.apache.seatunnel.engine.server.TaskExecutionService - [localhost]:5801 [seatunnel] [5.1] Exception in org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask@48afb905 java.util.concurrent.CompletionException: java.lang.ClassCastException: cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit at com.hazelcast.spi.impl.AbstractInvocationFuture.returnOrThrowWithJoinConventions(AbstractInvocationFuture.java:819) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.AbstractInvocationFuture.resolveAndThrowWithJoinConvention(AbstractInvocationFuture.java:835) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.AbstractInvocationFuture.join(AbstractInvocationFuture.java:553) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.engine.server.task.context.SeaTunnelSplitEnumeratorContext.assignSplit(SeaTunnelSplitEnumeratorContext.java:82) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplitEnumerator.lambda$assignSplit$11(KafkaSourceSplitEnumerator.java:305) ~[connector-kafka-2.3.3.jar:2.3.3] at java.util.HashMap.forEach(HashMap.java:1289) ~[?:1.8.0_292] at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplitEnumerator.assignSplit(KafkaSourceSplitEnumerator.java:303) ~[connector-kafka-2.3.3.jar:2.3.3] at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplitEnumerator.run(KafkaSourceSplitEnumerator.java:132) ~[connector-kafka-2.3.3.jar:2.3.3] at org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask.stateProcess(SourceSplitEnumeratorTask.java:303) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask.call(SourceSplitEnumeratorTask.java:134) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:613) ~[seatunnel-starter.jar:2.3.3] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_292] at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_292] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_292] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_292] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_292] Caused by: java.lang.ClassCastException: cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2301) ~[?:1.8.0_292] at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1431) ~[?:1.8.0_292] at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2410) ~[?:1.8.0_292] at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2328) ~[?:1.8.0_292] at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2186) ~[?:1.8.0_292] at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1666) ~[?:1.8.0_292] at java.io.ObjectInputStream.readObject(ObjectInputStream.java:502) ~[?:1.8.0_292] at java.io.ObjectInputStream.readObject(ObjectInputStream.java:460) ~[?:1.8.0_292] at org.apache.seatunnel.common.utils.SerializationUtils.deserialize(SerializationUtils.java:74) ~[connector-kudu-2.3.3.jar:2.3.3] at org.apache.seatunnel.api.serialization.DefaultSerializer.deserialize(DefaultSerializer.java:41) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.api.serialization.DefaultSerializer.deserialize(DefaultSerializer.java:25) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation.lambda$run$0(AssignSplitOperation.java:67) ~[seatunnel-starter.jar:2.3.3] at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48) ~[connector-kudu-2.3.3.jar:2.3.3] at org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation.run(AssignSplitOperation.java:54) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.Operation.call(Operation.java:189) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.call(OperationRunnerImpl.java:273) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:248) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:213) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:175) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:139) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123) ~[seatunnel-starter.jar:2.3.3] at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102) ~[seatunnel-starter.jar:2.3.3] 2024-01-09 20:05:15,789 INFO org.apache.seatunnel.engine.server.TaskExecutionService - [localhost]:5801 [seatunnel] [5.1] taskDone, taskId = 20000, taskGroup = TaskGroupLocation{jobId=797073430418554881, pipelineId=1, taskGroupId=1} 2024-01-09 20:05:15,789 INFO org.apache.seatunnel.engine.server.TaskExecutionService - [localhost]:5801 [seatunnel] [5.1] Task TaskGroupLocation{jobId=797073430418554881, pipelineId=1, taskGroupId=1} complete with state FAILED 2024-01-09 20:05:15,789 INFO org.apache.seatunnel.engine.server.CoordinatorService - [localhost]:5801 [seatunnel] [5.1] Received task end from execution TaskGroupLocation{jobId=797073430418554881, pipelineId=1, taskGroupId=1}, state FAILED 2024-01-09 20:05:15,790 INFO org.apache.kafka.common.utils.AppInfoParser - App info kafka.admin.client for seatunnel-enumerator-admin-client-1405936583 unregistered 2024-01-09 20:05:15,790 INFO org.apache.seatunnel.engine.server.dag.physical.PhysicalVertex - Job 12207549445376_job (797073430418554881), Pipeline: [(1/1)], task: [pipeline-1 [Source[0]-Kafka-Table12207303300864]-SplitEnumerator (1/1)] turn to end state FAILED. 2024-01-09 20:05:15,790 ERROR org.apache.seatunnel.engine.server.dag.physical.PhysicalVertex - Job 12207549445376_job (797073430418554881), Pipeline: [(1/1)], task: [pipeline-1 [Source[0]-Kafka-Table12207303300864]-SplitEnumerator (1/1)] end with state FAILED and Exception: java.util.concurrent.CompletionException: java.lang.ClassCastException: cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit at com.hazelcast.spi.impl.AbstractInvocationFuture.returnOrThrowWithJoinConventions(AbstractInvocationFuture.java:819) at com.hazelcast.spi.impl.AbstractInvocationFuture.resolveAndThrowWithJoinConvention(AbstractInvocationFuture.java:835) at com.hazelcast.spi.impl.AbstractInvocationFuture.join(AbstractInvocationFuture.java:553) at org.apache.seatunnel.engine.server.task.context.SeaTunnelSplitEnumeratorContext.assignSplit(SeaTunnelSplitEnumeratorContext.java:82) at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplitEnumerator.lambda$assignSplit$11(KafkaSourceSplitEnumerator.java:305) at java.util.HashMap.forEach(HashMap.java:1289) at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplitEnumerator.assignSplit(KafkaSourceSplitEnumerator.java:303) at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplitEnumerator.run(KafkaSourceSplitEnumerator.java:132) at org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask.stateProcess(SourceSplitEnumeratorTask.java:303) at org.apache.seatunnel.engine.server.task.SourceSplitEnumeratorTask.call(SourceSplitEnumeratorTask.java:134) at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:613) 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: java.lang.ClassCastException: cannot assign instance of org.apache.kafka.common.TopicPartition to field org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit.topicPartition of type org.apache.kafka.common.TopicPartition in instance of org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaSourceSplit at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(ObjectStreamClass.java:2301) at java.io.ObjectStreamClass.setObjFieldValues(ObjectStreamClass.java:1431) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2410) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2328) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2186) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1666) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:502) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:460) at org.apache.seatunnel.common.utils.SerializationUtils.deserialize(SerializationUtils.java:74) at org.apache.seatunnel.api.serialization.DefaultSerializer.deserialize(DefaultSerializer.java:41) at org.apache.seatunnel.api.serialization.DefaultSerializer.deserialize(DefaultSerializer.java:25) at org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation.lambda$run$0(AssignSplitOperation.java:67) at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48) at org.apache.seatunnel.engine.server.task.operation.source.AssignSplitOperation.run(AssignSplitOperation.java:54) at com.hazelcast.spi.impl.operationservice.Operation.call(Operation.java:189) at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.call(OperationRunnerImpl.java:273) at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:248) at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:213) at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:175) at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:139) at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123) at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102)

[github-actions[bot]]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.

[github-actions[bot]]

This issue has been closed because it has not received response for too long time. You could reopen it if you encountered similar problems in the future.

[weipengfei-sj]

version:2.3.5 mode:cluster mode I'm having the same issue