add kafka to bigtable blueprint by derrickaw · Pull Request #4363 · GoogleCloudPlatform/DataflowTemplates

derrickaw · 2026-10-03T21:19:48Z

add kafkaToBigtable blueprint IT etc
remove not supported parameters from kafka_options and other blueprints until supported

codecov · 2026-10-03T21:24:04Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 38.08%. Comparing base (5663e39) to head (5dd9b6d).
⚠️ Report is 16 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #4363      +/-   ##
============================================
+ Coverage     33.89%   38.08%   +4.19%     
- Complexity      505      517      +12     
============================================
  Files           215      195      -20     
  Lines         13186    11853    -1333     
  Branches       1320     1209     -111     
============================================
+ Hits           4469     4514      +45     
+ Misses         8350     6968    -1382     
- Partials        367      371       +4

Components	Coverage Δ
spanner-templates	`∅ <ø> (∅)`
spanner-import-export	`∅ <ø> (∅)`
spanner-live-forward-migration	`∅ <ø> (∅)`
spanner-live-reverse-replication	`∅ <ø> (∅)`
spanner-bulk-migration	`∅ <ø> (∅)`
gcs-spanner-dv	`∅ <ø> (∅)`
see 28 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

gemini-code-assist · 2026-10-04T00:03:49Z

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request introduces a new Kafka to Bigtable YAML template for Dataflow, facilitating seamless streaming data ingestion. It also performs a cleanup of existing Kafka-based templates by removing several unsupported configuration parameters, ensuring a more stable and consistent configuration interface across the template suite.

Highlights

New Template: Added a new Kafka to Bigtable YAML template to support streaming data ingestion from Kafka to Bigtable.
Integration Testing: Implemented comprehensive integration tests for the new Kafka to Bigtable template.
Parameter Cleanup: Removed unsupported Kafka configuration parameters (allowDuplicates, offsetDeduplication, redistributeByRecordKey, redistributeNumKeys, redistributed) from existing BigQuery and Iceberg templates.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

gemini-code-assist

Code Review

This pull request introduces the new Kafka to Bigtable (YAML) template, which includes its Java interface, YAML pipeline definition, documentation, and integration tests. Additionally, it cleans up several unused parameters (such as allowDuplicates and redistribution options) from existing Kafka-to-Iceberg and Kafka-to-BigQuery YAML templates. The review feedback highlights several improvement opportunities: managing the KafkaProducer in the integration test with a try-with-resources block to avoid resource leaks, correcting the YAML list syntax for requirements in KafkaToBigtable.yaml, and providing realistic examples instead of empty strings for the language and fields template parameters.

damccorm

Generally this LGTM. I did have one open question

damccorm · 2026-10-08T14:12:49Z

        required: true
        type: text
      - order: 3
-        name: "allowDuplicates"


Why are these options unsupported? Have they just not made it to yaml yet?

see apache/beam#40396, but they were not fully connected when initially added.

damccorm · 2026-10-08T14:20:26Z

Actually, found a few more potential issues with Jetski, I'll add those too.

damccorm · 2026-10-08T14:20:37Z

+    - type: WindowInto
+      name: WindowInto
+      input: ConvertStringsToBytes
+      config:
+        windowing: {{ windowing }}
+    - type: WriteToBigTable
+      name: WriteToBigtable
+      input: WindowInto
+      config:
+        project: {{ projectId }}
+        instance: {{ instanceId }}
+        table: {{ tableId }}


windowing is marked as optional = true in KafkaToBigtableYaml.java (because windowinto_options.yaml has is_optional: false instead of required: true or required: false), but WindowInto is included unconditionally here. If a user omits windowing, main.py strips the undefined windowing key and leaves WindowInto with config: {}, which will fail at pipeline construction.

Since Bigtable writes don't require windowing, should we make WindowInto conditional when windowing is provided (and update windowinto_options.yaml to use required: false)?

Suggested change

- type: WindowInto

name: WindowInto

input: ConvertStringsToBytes

config:

windowing: {{ windowing }}

- type: WriteToBigTable

name: WriteToBigtable

input: WindowInto

config:

project: {{ projectId }}

instance: {{ instanceId }}

table: {{ tableId }}

{% if windowing %}

- type: WindowInto

name: WindowInto

input: ConvertStringsToBytes

config:

windowing: {{ windowing }}

{% endif %}

- type: WriteToBigTable

name: WriteToBigtable

input: {% if windowing %}WindowInto{% else %}ConvertStringsToBytes{% endif %}

config:

project: {{ projectId }}

instance: {{ instanceId }}

table: {{ tableId }}

It should match the existing blueprint and be required. I have updated. Thanks

damccorm · 2026-10-08T14:20:38Z

+    - type: MapToFields
+      name: ConvertStringsToBytes
+      input: ReadFromKafka
+      config:
+        language: {{ language }}
+        fields: {{ fields }}


Question: Unlike PubSubToBigtable.yaml and KafkaToBigQuery.yaml, ReadFromKafka and MapToFields here don't configure error_handling with a dead-letter sink. Was omitting a DLQ intentional for the initial blueprint (similar to KafkaToIceberg.yaml), or should we include error handling so a malformed record / failed Python expression doesn't stall the streaming pipeline?

We should probably include this (and it should probably be included in KafkaToIceberg as well

Intentional move at this point to match existing. In future we can add DLQ to all if it makes sense to.

Could we take this as a fast follow then? I think it would be a good feature to have for all streaming blueprints.

Sure, after all blueprints are migrated and match JB, everything is supported in JB during reimport, JB examples are decommissioned, then I will follow up with DLQ. Sound good? Thanks

damccorm · 2026-10-08T14:20:38Z

-      example = "true")
-  Boolean getAllowDuplicates();
-
  @TemplateParameter.Text(


Since this PR cleans up unsupported parameters in kafka_options.yaml and fixes consumer_config: {{ consumerConfigUpdates }} in KafkaToIceberg.yaml, should we also wire confluent_schema_registry_subject, confluent_schema_registry_url, consumer_config, file_descriptor_path, and message_name into ReadFromKafka in KafkaToBigQuery.yaml? Right now KafkaToBigQueryYaml.java exposes those parameters from kafka_read_options, but KafkaToBigQuery.yaml only passes bootstrap_servers, topic, format, and schema.

I would prefer I open a separate PR to add supported parameters.

Opened #4397 to track these requests. Thanks

damccorm · 2026-10-08T14:20:38Z

        help: >
          The language used to define (and execute) the expressions and/or 
          callables in fields. Defaults to generic.
+        example: "python"
        required: true


Nit: The help text for language says "Defaults to generic.", but required: true is set without a default. Should this either have default: "generic" (with required: false), or should "Defaults to generic." be removed from the help text?

Thanks, updated

damccorm · 2026-10-08T14:20:38Z

+      order = 12,
+      name = "projectId",
+      optional = false,
+      description = "Bigtable project ID",
+      helpText = "The Google Cloud project ID of the Bigtable instance.",
+      example = "")
+  @Validation.Required
+  String getProjectId();
+
+  @TemplateParameter.Text(
+      order = 13,
+      name = "instanceId",
+      optional = false,
+      description = "Bigtable instance ID",
+      helpText = "The Bigtable instance ID.",
+      example = "")
+  @Validation.Required
+  String getInstanceId();
+
+  @TemplateParameter.Text(
+      order = 14,
+      name = "tableId",
+      optional = false,
+      description = "Bigtable output table",
+      helpText = "Bigtable table ID to write the output to.",
+      example = "")
+  @Validation.Required
+  String getTableId();
+
+  @TemplateParameter.Text(
+      order = 15,
+      name = "windowing",
+      optional = true,
+      description = "Windowing options",
+      helpText =
+          "Windowing options - see https://beam.apache.org/documentation/sdks/yaml/#windowing",
+      example = "")
+  String getWindowing();


Nit: projectId, instanceId, tableId (in bigtable_options.yaml) and windowing (in windowinto_options.yaml) all generate example = "". Consider adding example values in those option files as well.

updated, thanks

pull-request-size Bot added the size/XXL label Oct 3, 2026

derrickaw changed the title ~~20261003 kafka to big table~~ add kafka to bigtable blueprint Oct 3, 2026

derrickaw added the addition New feature or request label Oct 3, 2026

derrickaw marked this pull request as ready for review October 4, 2026 00:01

gemini-code-assist Bot reviewed Oct 4, 2026

View reviewed changes

derrickaw added 4 commits October 6, 2026 19:27

add kafkaToBT blueprint

0a3a69d

remove unused parameters until fully supported

40bf420

add examples to maptofields parameters, fix resource closure, etc

901e188

regen

1d0c24e

derrickaw force-pushed the 20261003_kafkaToBigTable branch from 0b215a2 to 1d0c24e Compare October 6, 2026 19:31

derrickaw requested a review from damccorm October 7, 2026 12:54

damccorm reviewed Oct 8, 2026

View reviewed changes

damccorm mentioned this pull request Oct 8, 2026

add kafka to spanner blueprint, IT etc #4365

Open

address reviewers' comments

5dd9b6d

derrickaw requested a review from damccorm October 8, 2026 16:16

Conversation

derrickaw commented Oct 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov Bot commented Oct 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

gemini-code-assist Bot commented Oct 4, 2026

Summary of Changes

Highlights

Footnotes

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

damccorm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

damccorm commented Oct 8, 2026

Uh oh!

damccorm Oct 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

derrickaw commented Oct 3, 2026 •

edited

Loading

codecov Bot commented Oct 3, 2026 •

edited

Loading

damccorm Oct 8, 2026 •

edited

Loading