Aiven-Open / Aiven-Open/bigquery-connector-for-apache-kafka
Supporting SMTs that change the topic name when upsertEnabled is set
- Langage dominant
- Java
- Étoiles
- 37
- Forks
- 45
- Merge moyen
- 19 h 50 min
- PR mergées (30 j)
- 5
Description
The original Confluent documentation that this project was forked from stated the following limitation: https://docs.confluent.io/kafka-connectors/bigquery/current/overview.html#limitations
> When the connector is configured with upsertEnabled or deleteEnabled, it does not support Single Message Transformations (SMTs) that modify the topic name. Additionally, the following transformations are not allowed: ... org.apache.kafka.connect.transforms.RegexRouter ....
From looking around the old GitHub issues for the original project from a few years ago, I got the impression that this was due to a fundamental Kafka Connect limitation that was subsequently fixed in [KIP-793: Allow sink connectors to be used with topic-mutating SMTs](https://cwiki.apache.org/confluence/display/KAFKA/KIP-793%3A+Allow+sink+connectors+to+be+used+with+topic-mutating+SMTs). According to KIP-793, connectors can subsequently gain compatibility with these SMTs by ensuring that the original pre-transformation offsets are used with the `preCommit` sink task function:
1. Connectors that examine offsets for `SinkRecord` instances provided to `put` function must use new functions `originalTopic`, `originalKafkaPartition`, and `originalKafkaOffset` if the results would later be returned to the `preCommit` function.
2. The `preCommit` function has to return the values from these new functions.
Examining the function at https://github.com/Aiven-Open/bigquery-connector-for-apache-kafka/blob/bc35e64fcde7cc3e227561a99102669289092fef/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/write/batch/MergeBatches.java#L361-L368 it does not look like these new functions are being used for the offsets returned by `preCommit` when flushing. (I found the same code as-is unmodified in Confluent's old connector.)
I did find a reference to the new Kafka 3.6 functions here: https://github.com/Aiven-Open/bigquery-connector-for-apache-kafka/blob/bc35e64fcde7cc3e227561a99102669289092fef/kcbq-connector/src/main/java/com/wepay/kafka/connect/bigquery/convert/KafkaDataBuilder.java#L163 but it only seems to be used for the output field populated by the `kafkaDataFieldName` config.
Guide de contribution
Aucun guide de contribution indexé pour ce dépôt
Évaluation
Cette issue n'a pas encore été évaluée.