BigQuery Connector

You want the messages already flowing through a Kafka topic to land in BigQuery
so your analysts can query them. This article walks through setting up the
Kafka Connect cluster that runs the connector, and the BigQuery sink connector
itself.

How it fits together

Two Components in SVIEW do the work. Kafka Connect (app-k8s-kafka-connect)
runs the worker pods; Kafka Connector (app-k8s-kafka-connector) holds the
individual connector definitions, such as the BigQuery sink.

The Connect workers authenticate to your Kafka brokers over TLS with SASL,
consume from the topics you grant them, and write each message into BigQuery
using a Google service account key that the platform mounts into the worker
pods for them. The Connect cluster is named after the Component instance —
kafka-connect-<instance>, defaulting to kafka-connect-shared.

Prerequisites

A Kafka cluster running in your environment.
A Kafka user with credentials and ACLs covering the topics you want to sink —
see Access Management.
A connector image containing the plugin JARs. The platform provides a default
image with the BigQuery sink connector already included, so you normally do
not need to build one.

1. Create the Google service account

The sink connector authenticates to BigQuery with a Google service account key.
Create the account, grant it the two BigQuery roles it needs, and download a
key file:

gcloud iam service-accounts create kafka-bq-sink \
  --display-name="Kafka BQ Sink Connector" \
  --project=<your-project-id>

gcloud projects add-iam-policy-binding <your-project-id> \
  --member="serviceAccount:kafka-bq-sink@<your-project-id>.iam.gserviceaccount.com" \
  --role="roles/bigquery.dataEditor"

gcloud projects add-iam-policy-binding <your-project-id> \
  --member="serviceAccount:kafka-bq-sink@<your-project-id>.iam.gserviceaccount.com" \
  --role="roles/bigquery.jobUser"

gcloud iam service-accounts keys create sa.json \
  --iam-account=kafka-bq-sink@<your-project-id>.iam.gserviceaccount.com

2. Store the key and inject it

On AWS-hosted Kafka clusters, store the service account key in AWS Secrets
Manager and reference it from the Kafka Connect Component's dependency.ssm
block. The platform takes care of turning that stored secret into a file the
Connect workers can read.

dependency:
  ssm:
    secretName: gcp-bq-sa-key
    json: "true"
    keys:
      - ssmKeys:
          - sa.json
        namespace: <kafka-namespace>
        sOverride: secretsmanager
        json: "false"

The key becomes available to the connector at
/opt/kafka/external-configuration/gcp-sa-key/SA.JSON. For how the ssm
dependency engine works in general, see
Application Dependencies.

3. Mount anything else through externalConfiguration

Kafka Connect on PushOps runs under the Strimzi operator, which limits how the
worker pods can be customised. Two limits are worth knowing before you write
your values:

Mount secrets into Connect pods with externalConfiguration.volumes. Using
template.pod.volumes instead is not valid for a Connect resource and will
make the deployment fail to sync.
template.connectContainer accepts only env and securityContext. Any
volumeMounts entries you add there are silently ignored.

Verifying it worked

Open the Kafka Connect Component in SVIEW and check that the worker pods are
healthy, then open the Kafka Connector Component and confirm the connector
reports a running state. The definitive confirmation is data arriving: query
the target BigQuery table and check that rows are appearing for recent
messages.

Deeper debugging

The commands below are for looking inside the Connect workers directly. They
require Kubernetes namespace access to the Kafka namespace — request it in
SVIEW under Access requests first, then connect through Teleport. If you do
not have that access, raise a request and include the connector name and the
symptom instead.

# Connect cluster status
kubectl get kafkaconnect -n <kafka-namespace>

# Connect worker logs
kubectl logs -n <kafka-namespace> kafka-connect-shared-connect-0

# Which connector plugins are registered
kubectl exec -n <kafka-namespace> kafka-connect-shared-connect-0 -- \
  curl -s http://localhost:8083/connector-plugins | python3 -m json.tool

# A connector's live configuration
kubectl exec -n <kafka-namespace> kafka-connect-shared-connect-0 -- \
  curl -s http://localhost:8083/connectors/<name>/config | python3 -m json.tool

# Turn on DEBUG logging for the BigQuery sink without restarting
kubectl exec -n <kafka-namespace> kafka-connect-shared-connect-0 -- \
  curl -X PUT http://localhost:8083/admin/loggers/com.wepay.kafka.connect.bigquery \
  -H "Content-Type: application/json" -d '{"level":"DEBUG"}'

Access Management
Debezium CDC Connector (GCP Cloud SQL PostgreSQL) — a source connector example
Application Dependencies