review
Code review a pull request for Redpanda Connect, checking Go patterns, tests, component architecture, and commit policy
Audit every consumer of the schema.Common metadata format (the format produced by schema_registry_decode's store_schema_metadata, the parquet_decode processor, and CDC sources) for type-coverage drift and value-coercion gaps. Run this whenever a new component starts consuming
$ npx -y skills add redpanda-data/connect --skill common-schema-audit --agent claude-codeHow it fires
How this skill gets triggered: by you, by Claude, or both.
/common-schema-auditContext preview
The summary Claude sees to decide when to auto-load this skill.
Audit every consumer of the schema.Common metadata format (the format produced by schema_registry_decode's store_schema_metadata, the parquet_decode processor, and CDC sources) for type-coverage drift and value-coercion gaps. Run this whenever a new component starts consuming
name: common-schema-audit description: Audit every consumer of the schema.Common metadata format (the format produced by schema_registry_decode's store_schema_metadata, the parquet_decode processor, and CDC sources) for type-coverage drift and value-coercion gaps. Run this whenever a new component starts consuming schema.Common, when a new schema.CommonType variant is added upstream in benthos, or as a periodic maintenance check. argument-hint: "[--format=md|json] [--component=<name>]" disable-model-invocation: true allowed-tools: Bash(go *), Bash(grep *), Bash(find *), Read, Glob, Grep, Task
`schema.Common` (from `github.com/redpanda-data/benthos/v4/public/schema`) is the canonical type metadata that flows through `meta(schema)` between Avro / Parquet / CDC sources and downstream sinks. Every consumer of this metadata must:
1. **Handle every variant of `schema.CommonType`** — or fail loudly with a useful error that names the missing case, not a generic "unsupported". 2. **Coerce values when the Go type of the message body doesn't match the schema-declared type** — specifically the temporal-to-numeric and numeric-to-temporal bridges that the iceberg shredder implements via `coerceTemporalToNumeric` and the metadata-aware path in `internal/impl/iceberg/shredder/temporal.go:208`.
This skill produces a per-consumer report so reviewers can catch drift before it ships.
The "GF iceberg issue" was a value-vs-metadata mismatch class. Fixes closed the gap in each consumer:
A new consumer of `schema.Common`, or a new `schema.CommonType` variant added upstream in benthos, can re-introduce the same bug class without anyone noticing until a customer pipeline breaks. The audit catches the drift mechanically.
1. **Enumerate the type universe.** Read every `schema.CommonType` constant from the benthos source — the authoritative list of variants every consumer must consider.
gopath=$(go env GOMODCACHE) benthos_dir=$(ls -d $gopath/github.com/redpanda-data/benthos/v4@*/ | tail -1) grep -E '^\s*(Boolean|Int32|Int64|Float32|Float64|String|ByteArray|Object|Map|Array|Null|Union|Timestamp|Date|TimeOfDay|UUID|Decimal|BigDecimal|Any)\s+CommonType' "$benthos_dir/public/schema/common.go"
Cross-check against the current set (as of the GF issue): `Boolean, Int32, Int64, Float32, Float64, String, ByteArray, Object, Map, Array, Null, Union, Timestamp, Date, TimeOfDay, UUID, Decimal, BigDecimal, Any`.
If new variants appear in benthos that aren't in this list, every consumer below will silently need an additional case — flag it loudly and update the skill's audit list.
2. **Find every consumer.** A "consumer" of `schema.Common` is a code path that reads parsed schema metadata and uses it to drive downstream type decisions. The reliable signal is a `schema.ParseFromAny(...)` call, plus any direct `schema.Common` type switches in encoding/coercion paths.
grep -rln 'schema\.ParseFromAny\|case schema\.\(Boolean\|Int32\|Int64\|Float32\|Float64\|String\|ByteArray\|Object\|Map\|Array\|Null\|Union\|Timestamp\|Date\|TimeOfDay\|UUID\|Decimal\|BigDecimal\|Any\)\b' internal/impl/ | grep -v _test
Producers (CDC schema builders in `mysql/`, `oracledb/`, `postgresql/`, `mongodb/cdc/`, `mssqlserver/`) are *not* consumers in this sense — they construct `schema.Common` from a source database's metadata; the type-coverage question doesn't apply. Filter those out.
3. **Per-consumer audit.** For each consumer, delegate to the Explore agent with the brief below. Run consumers in parallel.
Working dir: <connect repo>
Audit the consumer at <file>:<function> against the full schema.CommonType variant set:
Boolean, Int32, Int64, Float32, Float64, String, ByteArray, Object, Map, Array,
Null, Union, Timestamp, Date, TimeOfDay, UUID, Decimal, BigDecimal, Any.
Report:
(a) Type-coverage table: for each variant, which target type the consumer maps to (or whether it errors). Cite file:line.
(b) Value-coercion handling: when a message value's Go type doesn't match the
schema-declared type, does the consumer coerce or fail loudly? Specifically
check these cross-type cases:
- time.Time value + schema-declared Timestamp + integer-typed target column
- time.Duration value + schema-declared TimeOfDay + integer-typed target column
- Numeric int64 value + schema-declared Timestamp + integer-typed target column (unit-aware scaling)
- Numeric int32 value + schema-declared Date + integer-typed target column
Cite the coercion function and its location.
(c) Verdict: COVERED | PARTIAL | GAP, with one-line justification.
Reference implementations to compare against:
- iceberg shredder's coerceTemporalToNumeric in internal/impl/iceberg/shredder/temporal.go
- iceberg shredder's metadata-aware numeric scaling at temporal.go:208 onwards
- iceberg type_resolver's commonTypeToIcebergTypeRec in internal/impl/iceberg/type_resolver.go
Under 300 words per consumer.4. **Aggregate.** Combine the per-consumer reports into a single matrix:
| Consumer | Missing types | Missing coercions | Verdict | |---|---|---|---| | iceberg | (none) | (none) | COVERED | | parquet_encode | … | … | … | …
5. **Recommend.** For each GAP / PARTIAL row, propose the fix shape (port from iceberg, add cases to switch, etc.). Reference implementations to mirror, by file path so the pointers stay valid as the codebase evolves:
![Build Status][actions-url] ![Apache V2 API][godoc-url-apache] ![Enterprise API][godoc-url-enterprise] Redpanda Connect is a stream processor that moves data between a wide range of sources and sinks, with support for hydration, enrichment, transformation,
Code review a pull request for Redpanda Connect, checking Go patterns, tests, component architecture, and commit policy