[SUPPORT] kafka connect to Hudi - getting Transaction Participant doesn't exist error while trying to add Hudi-Sink to the connector

[yashps99]

Tips before filing an issue

• Have you gone through our FAQs?

• Join the mailing list to engage in conversations and get faster support at dev-subscribe@hudi.apache.org.

• If you have triaged this as a bug, then file an issue directly.

Describe the problem you faced

What could be the possible reason for this issue I'm getting while trying to add the hudi-sink to the connector?

To Reproduce

Steps to reproduce the behavior:

1. Zookeeper and Kafka running on separate cluster
2. Create the hudi-control-topic and hudi-test-topic
3. Go to the kafka home directory and run：./bin/connect-distributed.sh $HUDI_DIR/connect-distributed.properties
   4.Initiate a CONNECT request and execute：curl -X POST -H "Content-Type:application/json" -d @HUDI_DIR/config-sink.json http://localhost:8083/connectors

Environment Description

• Hudi version : 0.9.0

• Spark version : 2.4.8

• Hive version : 2.3.8

• Hadoop version : 2.10.1

• Storage (HDFS/S3/GCS..) : S3

• Running on Docker? (yes/no) : no

note：
1、connect-distributed.properties configuration：

bootstrap.servers=xx.xx.xx.xx:9092
group.id=hudi-connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
config.storage.topic=connect-configs
config.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1

offset.flush.interval.ms=60000
listeners=HTTP://:8083
plugin.path=/usr/local/share/kafka/plugins

2、config-sink-test.json configuration：

{
"name": "hudi-test-topic",
"config": {
"bootstrap.servers": xx.xx.xx.xx:9092",
"connector.class": "org.apache.hudi.connect.HoodieSinkConnector",
"tasks.max": "1",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter.schemas.enable": "false",
"topics": "hudi-test-topic",
"hoodie.table.name": "test_hudi_table",
"hoodie.table.type": "MERGE_ON_READ",
"hoodie.base.path": "s3a://",
"hoodie.datasource.write.partitionpath.field": "date",
"hoodie.datasource.write.recordkey.field": "volume",
"hoodie.kafka.commit.interval.secs": 60
}
}

Stacktrace

[2023-05-15 14:40:53,244] ERROR WorkerSinkTask{id=hudi-sink-0} RetriableException from SinkTask: (org.apache.kafka.connect.runtime.WorkerSinkTask:601)
org.apache.kafka.connect.errors.RetriableException: TransactionParticipant should be created for each assigned partition, but has not been created for the topic/partition: hudi-test-topic:0
	at org.apache.hudi.connect.HoodieSinkTask.put(HoodieSinkTask.java:111)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:582)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:330)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:232)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:188)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:237)
	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)

Refer to the link：
https://github.com/apache/hudi/tree/master/hudi-kafka-connect

[ad1happy2go]

@yashps99 Can you try with a later version i.e. 0.12.2. I did tried to setup demo Kafka connect and it all worked fine. Also did you created the topic before starting the connect.

[yashps99]

@ad1happy2go I can't change the version as it's managed by organisation. And yes I've created the topic before starting the connect. And the hudi version is - 0.9.0.

[yashps99]

@ad1happy2go does hudi version pre-0.10.0 doesn't support the kafka-connect sink approach?

[ad1happy2go]

@yashps99 Actually I am not able to use that version in my setup, So unable to support with version 0.9.0.
I can confirm later versions are definitely working with Kafka-connect.

[xushiyan]

@yashps99 Did you able to get it working with later hudi version? Or its not possible. 0.9.0 is anyway out of support. So ideally you should try to get it upgraded.

[ad1happy2go]

@yashps99 Any updates on this?

[Rajvir-Singh]

Add below properties to config-sink-test.json

"hoodie.schemaprovider.class": "org.apache.hudi.schema.FilebasedSchemaProvider",
"hoodie.streamer.schemaprovider.source.schema.file":"file:///C:/Users/abc/Desktop/user_schema.avsc",

[rangareddy]

Hi @yashps99

I think you did not share the full exception. To find out the cause, you need to check the complete exception, similar to the one below:

[2024-12-06 07:14:39,980] WARN Failed to initialize filesystem s3a:///warehouse/hudi-test-topic: java.lang.NullPointerException: null uri host. (org.apache.hadoop.fs.FileSystem)
[2024-12-06 07:14:39,984] ERROR Fatal error initializing task null for partition 0 (org.apache.hudi.connect.HoodieSinkTask)
org.apache.hudi.exception.HoodieException: Fatal error instantiating Hudi Transaction Services 
	at org.apache.hudi.connect.writers.KafkaConnectTransactionServices.<init>(KafkaConnectTransactionServices.java:109)
	at org.apache.hudi.connect.transaction.ConnectTransactionCoordinator.<init>(ConnectTransactionCoordinator.java:88)
	at org.apache.hudi.connect.HoodieSinkTask.bootstrap(HoodieSinkTask.java:191)
	at org.apache.hudi.connect.HoodieSinkTask.open(HoodieSinkTask.java:151)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.openPartitions(WorkerSinkTask.java:651)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.access$1200(WorkerSinkTask.java:74)
	at org.apache.kafka.connect.runtime.WorkerSinkTask$HandleRebalance.onPartitionsAssigned(WorkerSinkTask.java:748)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsAssigned(ConsumerCoordinator.java:322)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:471)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:474)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:385)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:557)
	at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1272)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1236)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1216)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.pollConsumer(WorkerSinkTask.java:480)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:332)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:238)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:207)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:213)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:268)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
	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.hudi.exception.HoodieException: Unable to create org.apache.hudi.storage.hadoop.HoodieHadoopStorage
	at org.apache.hudi.storage.HoodieStorageUtils.getStorage(HoodieStorageUtils.java:44)
	at org.apache.hudi.storage.HoodieStorageUtils.getStorage(HoodieStorageUtils.java:34)
	at org.apache.hudi.common.table.HoodieTableMetaClient.initTableAndGetMetaClient(HoodieTableMetaClient.java:466)
	at org.apache.hudi.common.table.HoodieTableMetaClient$PropertyBuilder.initTable(HoodieTableMetaClient.java:1199)
	at org.apache.hudi.connect.writers.KafkaConnectTransactionServices.<init>(KafkaConnectTransactionServices.java:105)
	... 26 more
Caused by: org.apache.hudi.exception.HoodieException: Unable to instantiate class org.apache.hudi.storage.hadoop.HoodieHadoopStorage
	at org.apache.hudi.common.util.ReflectionUtils.loadClass(ReflectionUtils.java:75)
	at org.apache.hudi.storage.HoodieStorageUtils.getStorage(HoodieStorageUtils.java:41)
	... 30 more
Caused by: java.lang.reflect.InvocationTargetException
	at jdk.internal.reflect.GeneratedConstructorAccessor57.newInstance(Unknown Source)
	at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:490)
	at org.apache.hudi.common.util.ReflectionUtils.loadClass(ReflectionUtils.java:73)
	... 31 more
Caused by: java.lang.NullPointerException: null uri host.
	at java.base/java.util.Objects.requireNonNull(Objects.java:248)
	at org.apache.hadoop.fs.s3native.S3xLoginHelper.buildFSURI(S3xLoginHelper.java:72)
	at org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:209)
	at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3611)
	at org.apache.hadoop.fs.FileSystem.access$300(FileSystem.java:174)
	at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3712)
	at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3663)

[rangareddy]

To write data to S3 using the Hudi Sink Connector, we need to pass additional S3-related parameters, which are not present in your connector configuration file.

I am sharing sample S3 configuration file content here.

{
  "name": "hudi-sink-connector-s3",
  "config": {
    "bootstrap.servers": "kafka:29092",
    "connector.class": "org.apache.hudi.connect.HoodieSinkConnector",
    "tasks.max": "1",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter.schemas.enable": "false",
    "topics": "hudi-test-topic",
    "hoodie.table.name": "hudi_test_table",
    "hoodie.table.type": "MERGE_ON_READ",
    "hoodie.base.path": "s3a://warehouse/hudi_test_table/",
    "fs.s3a.connection.ssl.enabled": "false",
    "fs.s3a.endpoint": "http://minio:9000",
    "fs.s3a.access.key": "admin",
    "fs.s3a.secret.key": "password",
    "fs.s3a.path.style.access": "true",
    "fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",
    "fs.s3a.aws.credentials.provider": "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider",
    "fs.s3a.endpoint.region": "us-east-1",
    "hoodie.datasource.write.recordkey.field": "volume",
    "hoodie.datasource.write.partitionpath.field": "date",
    "hoodie.schemaprovider.class": "org.apache.hudi.schema.SchemaRegistryProvider",
    "hoodie.streamer.schemaprovider.registry.url": "http://kafka-schema-registry:8081/subjects/hudi-test-topic/versions/latest",
    "hoodie.kafka.commit.interval.secs": 60
  }
}

Please test using the above configuration by modifying the required parameter values, and let me know if you need any further assistance.