...

TransactionParticipant class implements the processing of records per partition. Since in this case, Task0 is responsible for partitions 0 and 2, the HudiSinkTask will instantiate two instances of TransactionParticipant that will each handle a single partition. Each participant maintains a buffer for incoming Kafka records and an event queue that is used by KafkaCommunicationAgent to asynchronously submit control messages received from the Kafka control topic. The state machine manager represents the state of each participant based on events received from the coordinator via the control topic. Finally, the HudiJavaWriter is the java based implementation of the Hudi Writer that will read records from the incoming buffer and append the records to the Hudi File group.

Connect Sink Task

The HudiSinkTask pseudo code with key interface implementations is shown below. On initialization or when a new partition is assigned to the task (OPEN API), a new instance of TransactionParticipant is assigned to the partition. If the task manages partition 0, then an instance of TransactionCoordinator is also created, which runs continuously in a separate thread.

Kafka connect calls the PUT call periodically with records read from Kafka. First, the records are added to the buffer of the corresponding TransactionParticipant based on the partition. Then we execute the processRecords method of TransactionParticipant that runs the state machine followed by writing of the records to Hudi files. The implementation of TransactionParticipant is explained in more detail in the next section.

Kafka connect calls the PRECOMMIT periodically (the interval can be configured by user) to commit the last Kafka offset. The offset is returned by TransactionParticipant , which is updated based on the written records that were committed. Since Kafka offsets are committed as Hudi files are committed, we suggest setting the interval for PRECOMMIT similar to the transaction intervals.

```

class HudiSinkTask implements SinkTask void OPEN(partitions) { for (partition: partitions) { if (partition == 0) { transactionCoordinator.start() } transactionParticipant.get(partition).start() } } void PUT(records) { for (record : records) { transactionParticipant.get(record.partition).buffer(record) } for (partition: assigned_partitions) { transactionParticipant.get(record.partition).processRecords } } void PRECOMMIT() { commitOffsets = [] for (partition: assigned_partitions) { commitOffsets.put(partition, transactionParticipant.get(record.partition).getLatestKafkaOffset()) } return commitOffset }

```

Implementation

Rollout/Adoption Plan

...