[Bug] [connector-kafka] Class org.apache.kafka.common.serialization.ByteArraySerializer could not be found
- Dominant language
- Java
- Stars
- 9.7k
- Forks
- 2.4k
- Avg merge
- 3d 17h
- Merged PRs (30d)
- 210
Description
### Search before asking
- [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues.
### What happened
I use seatunnel to synchronize MySQL-cdc data to Kafka. An error occurs during task execution.
```java
org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.ByteArraySerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.ByteArraySerializer could not be found.
at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:744)
at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:490)
at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:483)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:113)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:133)
at org.apache.kafka.clients.producer.ProducerConfig.(ProducerConfig.java:553)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:289)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:316)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:301)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaInternalProducer.(KafkaInternalProducer.java:46)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.getProducer(KafkaSinkCommitter.java:88)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.commit(KafkaSinkCommitter.java:53)
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.notifyCheckpointComplete(SinkFlowLifeCycle.java:309)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyCheckpointComplete$8(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.common.utils.ExceptionUtil.sneaky(ExceptionUtil.java:138)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyAllAction$13(SeaTunnelTask.java:380)
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyAllAction(SeaTunnelTask.java:380)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyCheckpointComplete(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.lambda$runInternal$0(CheckpointFinishedOperation.java:93)
at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.runInternal(CheckpointFinishedOperation.java:81)
at org.apache.seatunnel.engine.server.task.operation.TracingOperation.run(TracingOperation.java:42)
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:471)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:197)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:137)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123)
at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102)
org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException: org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.ByteArraySerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.ByteArraySerializer could not be found.
at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:744)
at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:490)
at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:483)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:113)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:133)
at org.apache.kafka.clients.producer.ProducerConfig.(ProducerConfig.java:553)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:289)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:316)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:301)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaInternalProducer.(KafkaInternalProducer.java:46)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.getProducer(KafkaSinkCommitter.java:88)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.commit(KafkaSinkCommitter.java:53)
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.notifyCheckpointComplete(SinkFlowLifeCycle.java:309)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyCheckpointComplete$8(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.common.utils.ExceptionUtil.sneaky(ExceptionUtil.java:138)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyAllAction$13(SeaTunnelTask.java:380)
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyAllAction(SeaTunnelTask.java:380)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyCheckpointComplete(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.lambda$runInternal$0(CheckpointFinishedOperation.java:93)
at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.runInternal(CheckpointFinishedOperation.java:81)
at org.apache.seatunnel.engine.server.task.operation.TracingOperation.run(TracingOperation.java:42)
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:471)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:197)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:137)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123)
at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.lambda$runInternal$0(CheckpointFinishedOperation.java:99) ~[seatunnel-starter.jar:2.3.11]
at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48) ~[seatunnel-starter.jar:2.3.11]
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.runInternal(CheckpointFinishedOperation.java:81) ~[seatunnel-starter.jar:2.3.11]
at org.apache.seatunnel.engine.server.task.operation.TracingOperation.run(TracingOperation.java:42) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.Operation.call(Operation.java:189) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.call(OperationRunnerImpl.java:273) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:248) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:471) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:197) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:137) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102) ~[seatunnel-starter.jar:2.3.11]
[] 2025-09-26 15:13:19,350 ERROR [.o.CheckpointFinishedOperation] [hz.main.generic-operation.thread-27] - [10.233.83.113]:5801 [seatunnel-cluster] [5.1] org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.ByteArraySerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.ByteArraySerializer could not be found.
```
### SeaTunnel Version
2.3.11
### SeaTunnel Config
```conf
env {
# 环境配置
parallelism = 8 # 增加并行度以提高吞吐量
job.mode = "STREAMING" # 使用流式处理模式进行实时同步
checkpoint.interval = 10000 # 检查点间隔，单位毫秒
# 限流配置 - 避免对源数据库造成过大压力
read_limit.bytes_per_second = 10000000 # 每秒读取字节数限制，约10MB/s
read_limit.rows_per_second = 1000 # 每秒读取行数限制
# 本地检查点配置
execution.checkpoint.data-uri = "file:///opt/seatunnel/checkpoints"
execution.checkpoint.max-concurrent = 1 # 最大并发检查点数
# 性能优化参数
execution.buffer-timeout = 5000 # 缓冲超时时间(毫秒)
execution.jvm-options = "-Xms4g -Xmx8g -XX:+UseG1GC -XX:MaxGCPauseMillis=100"
}
source {
MySQL-CDC {
# 基本连接配置
server-id = 5652-5662 # MySQL复制客户端的唯一ID范围
base-url = "jdbc:mysql://xxxxm:3306/sp?useUnicode=true&characterEncoding=UTF-8&connectionTimezone=Asia/Shanghai&useSSL=false&rewriteBatchedStatements=true&useServerPrepStmts=true&cachePrepStmts=true"
username = "root"
password = "xxxx"
table-names = ["sp.user"] # 要同步的表
# CDC 特有配置
schema-changes.enabled = true # 启用架构变更捕获
connection-time-zone = "Asia/Shanghai" # 服务器时区
# 性能优化配置
snapshot.mode = "initial" # 初始快照模式
snapshot.fetch.size = 10000 # 快照获取大小
chunk.size.rows = 8096 # 分块大小，用于并行快照
connection.pool.size = 10 # 连接池大小
# 高级配置
include.schema.changes = true # 包含架构变更事件
scan.startup.mode = "initial" # 启动模式：initial(全量+增量)
scan.incremental.snapshot.chunk.size = 8096 # 增量快照分块大小
debezium.min.row.count.to.stream.results = 1000 # 流式结果的最小行数
# 容错配置
connect.timeout = 30000 # 连接超时时间(毫秒)
connect.max-retries = 3 # 最大重试次数
# 输出表名
result_table_name = "mysql_user_cdc_source"
}
}
# 可选的转换逻辑，如果需要对数据进行处理
transform {
Sql {
source_table_name = "mysql_user_cdc_source"
result_table_name = "kafka_sink_data"
# 根据需要转换字段，这里添加了一个分区字段k1
query = """
SELECT 's' AS 'application_code', channel_id AS 'channel_code', id AS 'user_id', phone AS 'mobile', real_name AS 'name', account, created_time AS 'register_time' FROM mysql_user_cdc_source
"""
}
}
sink {
kafka {
topic = "user_topic"
bootstrap.servers = "192.168.109.211:9092,192.168.109.212:9092,192.168.109.213:9092"
format = json
kafka.request.timeout.ms = 60000
kafka.config = {
acks = "all"
request.timeout.ms = 60000
buffer.memory = 33554432
}
}
}
```
### Running Command
```shell
helm seatunnel-on-k8s
bin/seatunnel.sh --config config/ods_user_cdc.conf -m cluster -n ods_user_cdc
```
### Error Exception
```log
org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.ByteArraySerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.ByteArraySerializer could not be found.
at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:744)
at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:490)
at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:483)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:113)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:133)
at org.apache.kafka.clients.producer.ProducerConfig.(ProducerConfig.java:553)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:289)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:316)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:301)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaInternalProducer.(KafkaInternalProducer.java:46)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.getProducer(KafkaSinkCommitter.java:88)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.commit(KafkaSinkCommitter.java:53)
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.notifyCheckpointComplete(SinkFlowLifeCycle.java:309)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyCheckpointComplete$8(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.common.utils.ExceptionUtil.sneaky(ExceptionUtil.java:138)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyAllAction$13(SeaTunnelTask.java:380)
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyAllAction(SeaTunnelTask.java:380)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyCheckpointComplete(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.lambda$runInternal$0(CheckpointFinishedOperation.java:93)
at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.runInternal(CheckpointFinishedOperation.java:81)
at org.apache.seatunnel.engine.server.task.operation.TracingOperation.run(TracingOperation.java:42)
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:471)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:197)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:137)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123)
at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102)
org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException: org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.ByteArraySerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.ByteArraySerializer could not be found.
at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:744)
at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:490)
at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:483)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:113)
at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:133)
at org.apache.kafka.clients.producer.ProducerConfig.(ProducerConfig.java:553)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:289)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:316)
at org.apache.kafka.clients.producer.KafkaProducer.(KafkaProducer.java:301)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaInternalProducer.(KafkaInternalProducer.java:46)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.getProducer(KafkaSinkCommitter.java:88)
at org.apache.seatunnel.connectors.seatunnel.kafka.sink.KafkaSinkCommitter.commit(KafkaSinkCommitter.java:53)
at org.apache.seatunnel.engine.server.task.flow.SinkFlowLifeCycle.notifyCheckpointComplete(SinkFlowLifeCycle.java:309)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyCheckpointComplete$8(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.common.utils.ExceptionUtil.sneaky(ExceptionUtil.java:138)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.lambda$notifyAllAction$13(SeaTunnelTask.java:380)
at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyAllAction(SeaTunnelTask.java:380)
at org.apache.seatunnel.engine.server.task.SeaTunnelTask.notifyCheckpointComplete(SeaTunnelTask.java:360)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.lambda$runInternal$0(CheckpointFinishedOperation.java:93)
at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.runInternal(CheckpointFinishedOperation.java:81)
at org.apache.seatunnel.engine.server.task.operation.TracingOperation.run(TracingOperation.java:42)
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:471)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:197)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:137)
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123)
at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102)
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.lambda$runInternal$0(CheckpointFinishedOperation.java:99) ~[seatunnel-starter.jar:2.3.11]
at org.apache.seatunnel.common.utils.RetryUtils.retryWithException(RetryUtils.java:48) ~[seatunnel-starter.jar:2.3.11]
at org.apache.seatunnel.engine.server.checkpoint.operation.CheckpointFinishedOperation.runInternal(CheckpointFinishedOperation.java:81) ~[seatunnel-starter.jar:2.3.11]
at org.apache.seatunnel.engine.server.task.operation.TracingOperation.run(TracingOperation.java:42) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.Operation.call(Operation.java:189) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.call(OperationRunnerImpl.java:273) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:248) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationservice.impl.OperationRunnerImpl.run(OperationRunnerImpl.java:471) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:197) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.process(OperationThread.java:137) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.spi.impl.operationexecutor.impl.OperationThread.executeRun(OperationThread.java:123) ~[seatunnel-starter.jar:2.3.11]
at com.hazelcast.internal.util.executor.HazelcastManagedThread.run(HazelcastManagedThread.java:102) ~[seatunnel-starter.jar:2.3.11]
[] 2025-09-26 15:13:19,350 ERROR [.o.CheckpointFinishedOperation] [hz.main.generic-operation.thread-27] - [10.233.83.113]:5801 [seatunnel-cluster] [5.1] org.apache.kafka.common.config.ConfigException: Invalid value org.apache.kafka.common.serialization.ByteArraySerializer for configuration key.serializer: Class org.apache.kafka.common.serialization.ByteArraySerializer could not be found.
```
### Zeta or Flink or Spark Version
zeta 2.3.11
### Java or Scala Version
jdk8
### Screenshots
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)
Contributor guide
No contributing guide indexed for this repository
Research direction
Start at connectors/connector-seatunnel-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/sink/KafkaInternalProducer.java:46 and trace how KafkaSinkCommitter creates the producer. Inspect the connector's dependency and class-loading setup for the missing ByteArraySerializer, then verify that a MySQL-CDC to Kafka streaming job can complete its checkpoint commit without the reported error.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka, mysql
- Domain
- data-engineering, databases, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100