Config enableBatchLoad does not work correctly when original topic name was mutated in SMT

[SamoylovMD]

Following my previous PR, when enableBatchLoad config is set, it doesn't work correctly when the topic name was mutated in SinkRecord.

I will create a separate PR if I have time to work on this. If you have any thoughts on implementation (for example, should we make the same handle for SinkRecord.originalTopic or approach it differently), please let me know. Ref to the problematic code is here.

By the way, what is the use case for having enableBatchLoad as list? As a user, I would expect this to be a boolean value, applied to all topics handled by the connector, not a subset - that would be much more convenient.

[Claudenw]

Is this a case where we need to check if the name was mutated and get the old name if tracked?

[SamoylovMD]

Yes, correct, when topic name is mutated in SMT, it kinda breaks enableBatchLoad setting, so you need to put mutated topics, not original ones.

My use case is reading from many topics and collecting all records in one target table, by rewriting topic name in each record. I have two connectors, one is plain streaming and works fine, the other one has enableBatchLoad, where I need to specify mutated topic name.

In the original connector, this setting meant a subset of topics which use batch data load through GCS instead of streaming inserts. While there is an option to have a wildcard style of source topics via topics.regex, enableBatchLoad needs a specific list of topics and doesn't support regex.

Maybe we can reimagine this in the new major versions of connector plugin.

[Claudenw]

Can you envision a solution? If so implement it with a new parameter. we can then put that in for a point release and deprecate the old setting and solution to be removed in a major version.

[Claudenw]

How about a switch to enable using original topic,partition, and offset or the mutated one?

[SamoylovMD]

@Claudenw sorry for missing the discussion around this.

I'm fine with having a switch, but enableBatchLoad should be consistent with it. After the patch #97, regardless of the switch between original and mutation-aware behaviour, the option enableBatchLoad acts always as if the switch is in 'original' position.

In my opinion, enableBatchLoad in its current semantic (which I don't like #93) should check record.originalTopic() if the switch that you implemented in #97 is true. That means enableBatchLoad should be kind of the same as we specify in topics or topics.regex and which is provisioned as metadata.topic, regardless of how it's changed in upstream SMT.

I also might be wrong if enableBatchLoad has a different semantic. If it's supposed to correspond to target table names, then it's better to document this behaviour somehow. But in this case the connector configuration should know how topic name was changed, which may be not a good idea.

[Claudenw]

@SamoylovMD I think I understand the issue.

I want to first state that I have only been working on this project for a couple of months. But my understanding of the batchLoad process is that it writes data to files on GCS and then uploads them into BigQuery.

So the process is something like:
receive record -> mutate -> collect into batch -> write batch to BigQuery table

@SamoylovMD is your problem that the 'collect into batch' is collecting based on the mutated topic and you want it to check on the original topic? Or is it that while the batch is collected based on the mutated topic, the record metadata written into the BigQuery record does not contain the original topic?

@RyanSkraba does this make sense to you? Would it make sense to look at the collector like the grouper in the cloud storage connector?

[SamoylovMD]

@Claudenw let me provide an example.

Here is a configuration of my connector:

{
  "name": "my-test-connector",
  "config": {
    "name": "my-test-connector",
    "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector",
    "tasks.max": "1",

    "keyfile": "/path/to/credentials.json",
    "project": "my-gcp-project",
    "defaultDataset": "my_dataset",

    "topics": "topic1,topic2,topic3",
    "enableBatchLoad": "topic1,topic2,topic3",
    "batchLoadIntervalSec": 60,
    "gcsBucketName": "my-gcp-project-bucket",

    "autoUpdateSchemas": "true",
    "autoCreateTables": "true",
    "sanitizeTopics": "true",
    "allowSchemaUnionization": "true",
    "allowBigQueryRequiredFieldRelaxation": "true",
    "kafkaDataFieldName": "metadata",
    "preserveKafkaTopicPartitionOffset": "true",

    "transforms": "myCoolTransformation",
    "transforms.myCoolTransformation.type": "<path to SMT class>"
  }
}

So, with the new behaviour (enabled by setting preserveKafkaTopicPartitionOffset=true if I understand correctly) the option enableBatchLoad is ignored if myCoolTransformation rewrites record topics. This is the problem I want to address.

I see the two valid options how we can handle this:

• make enableBatchLoad aware of the option preserveKafkaTopicPartitionOffset, so enableBatchLoad is consistent with topics in connector config;
• rename setting (or add new setting for backward compatibility) which specifically mentions target table, something like enableBatchLoadForTargetTables (name is discussable).

[Claudenw]

I understand your first option -- and that makes sense to me as the default as batch loading may, or may not, want the SMT values for table distribution.

I don't understand what the second option would do.

[SamoylovMD]

Sorry, I meant that the current name for the setting is too vague, and if we intentionally keep the current logic (which is, enabling batch load for specific sublist of resulting tables after SMT renaming), it's probably better to rename the setting or document the semantics.

I'm not advocating for this, just mentioning the possible solution if the users say it's crucial to keep the current logic around batch load, at least we should make it clearer how to use it correctly.