kparisa opened a new issue, #4393:
URL: https://github.com/apache/iggy/issues/4393
### Description
Add a Google BigQuery sink connector (`core/connectors/sinks/bigquery_sink`)
that writes messages from Iggy topics into BigQuery tables using the **BigQuery
Storage Write API**.
## Scope
- Write via the Storage Write API `_default` stream using Arrow-encoded
rows. Do not use the legacy `insertAll` API.
- Authentication:
- Service account key from a file or inline secret.
- Google Application Default Credentials (ADC).
- Configuration:
- `project_id`
- `dataset`
- `table`
- retries/backoff
- timeout
- maximum request size
- Support two ingestion modes:
- **`mapped`**: JSON payload fields map to BigQuery table columns. Fetch
and validate the target table schema during `open()`.
- **`raw`**: Store the original payload in a single column alongside
optional Iggy metadata.
- Optional Iggy metadata columns:
- `iggy_stream`
- `iggy_topic`
- `iggy_partition_id`
- `iggy_offset`
- `iggy_timestamp`
- `iggy_id`
- `iggy_headers`
- Payload support:
- JSON
- Text
- Raw payloads in `raw` mode
- JSON representations of Protobuf payloads where applicable
- In `mapped` mode, Text payloads are parsed as JSON. Rows that are not
valid JSON objects are rejected.
- Columns missing from a JSON payload use BigQuery's `DEFAULT_VALUE`
missing-value interpretation by default, with `NULL_VALUE` configurable where
appropriate.
- Split large batches below the Storage Write API `AppendRows` request
limit, using a conservative threshold such as ~8 MB.
- No buffering across `consume()` calls.
- Unsupported BigQuery column types such as `INTERVAL` and `RANGE` should
fail validation during `open()`.
## Error handling
Retry transient/retryable gRPC failures with bounded jittered backoff,
including:
- `UNAVAILABLE`
- `DEADLINE_EXCEEDED`
- `INTERNAL`
- `ABORTED`
- `RESOURCE_EXHAUSTED`
Treat errors such as the following as permanent:
- `INVALID_ARGUMENT`
- `PERMISSION_DENIED`
- `UNAUTHENTICATED`
For row-level errors, identify the offending rows, drop them, and re-append
the remaining valid rows once. Every dropped row must be logged with its
stream, topic, partition, and offset and counted in a connector metric.
There is no dead-letter queue in v1.
## Delivery guarantees
The connector runtime commits consumer group offsets when messages are
polled, before `consume()` runs. A batch for which `consume()` returns an error
is logged and not redelivered.
Consequences:
- The sink retries retryable BigQuery failures internally with bounded
backoff.
- A batch that still fails after the final retry, or encounters a permanent
error, is lost.
- A process crash between the consumer group offset commit and a successful
BigQuery append also loses that batch.
- With the `_default` stream, an ambiguous append result followed by a retry
can produce duplicates. The `iggy_stream`, `iggy_topic`, `iggy_partition_id`,
and `iggy_offset` metadata can be used for downstream deduplication.
Crash-safe delivery guarantees, whether at-least-once or exactly-once,
require a runtime change so consumer group offsets are committed only after the
sink successfully persists the batch. This should be tracked separately.
Future work may use committed write streams with explicit append offsets for
stronger in-process delivery guarantees.
## Implementation
- Use `gcloud-bigquery` (Apache-2.0) for the Storage Write API, table
metadata, and authentication. It aligns with workspace dependencies including
`arrow`, `reqwest`, and `prost`.
- `tonic` (with the `server` feature) is added as a dev-dependency of the
integration test crate for the fake BigQueryWrite service. The sink itself
reaches gRPC through `gcloud-bigquery`; `tonic` is already present transitively
in `Cargo.lock`.
- Follow existing sink connector conventions, including the
`clickhouse_sink` layout, SDK retry helpers, secrets kept out of logs, and
atomic counters.
## Testing
### Unit tests
Cover:
- Configuration and validation
- Schema mapping
- Arrow encoding
- Request-size splitting
- Error classification
- Retry behavior
- Row-level error handling
- Credential source selection:
- key file
- inline key
- ADC fallback
- missing/invalid key file
### Integration tests
Use an in-process fake BigQuery service consisting of a `tonic`
BigQueryWrite service and HTTP `tables.get` stub.
Cover:
- Successful writes
- Batch splitting
- Row-level errors
- Retryable failures
- Permanent failures
- `PERMISSION_DENIED` and `UNAUTHENTICATED` responses treated as permanent
errors
- Schema validation failures during `open()`
A real BigQuery dependency should not be required for CI. Document manual
verification against Google BigQuery in the connector README.
## Documentation
Add:
- Connector README
- Configuration reference
- Required Google Cloud IAM permissions
- Authentication examples
- BigQuery/Iggy type mapping
- Delivery guarantees and failure semantics
- Manual testing instructions
- Sample TOML under `runtime/example_config/connectors/`
- BigQuery entry in the sinks README
## Non-goals for v1
- Automatic table creation
- Automatic schema evolution
- Per-topic table routing
- CDC upserts/deletes using `_CHANGE_TYPE`
- Crash-safe delivery guarantees (requires a connector runtime change)
- Dead-letter queue for rejected rows
### Affected area / component
_No response_
### Proposed solution
_No response_
### Alternatives considered
_No response_
### Contribution
- [ ] I'm willing to submit a pull request to implement this feature
### Good first issue
- [ ] I think this could be a good first issue for a new contributor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
