Kafka Connect converter - DO NOT MERGE#4548
Open
prakhargarg105 wants to merge 73 commits into
Open
Conversation
Approved brainstorming design for a deterministic Go conversion engine (internal/connect_converter) plus thin CLI. v1 covers mirror, Snowflake, BigQuery, S3, GCS, and JDBC connectors, common converters and SMTs, with best-effort YAML output and inline # TODO markers for unmapped content. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
15-task TDD plan for internal/connect_converter engine + CLI convert subcommand. Covers parser, MapCtx, yaml builders, registry/fallback, assemble/render, six connector mappers, converters, SMTs, golden tests, and CLI wiring. Each task validates output via the benthos linter. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…eton Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… assemble invariant note Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…sistently Every required snowflake_streaming field now emits a TODO stub *and* calls ctx.Warn when the corresponding KC key is absent, fixing the silent drop of `user` and the missing Warn calls on `account`/`role`/`database`/`schema`. Optional fields (private_key, private_key_file) are mapped-only with no stub. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…outer) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Rename unused interface-method params to _, add const-block doc comment, apply SplitSeq modernize fix, and use assert.Empty for blank-string checks. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ld value coercion Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Usage (CLI + getting the Kafka Connect JSON via the Connect REST API), supported connectors/converters/SMTs, output model, v1 limitations, architecture, how to add a mapper, and the linter-backed test approach. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Adds design for ExtractField, HoistField, MaskField, Cast, TimestampConverter, ValueToKey SMT mappers. Purely additive to the existing SMTMapper/registry pattern. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…imestampConverter/ValueToKey SMTs Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…class Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…or aliases Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Strip // line comments (JSONC-style) before JSON unmarshaling so users can paste documented configs. Stripping is string-aware: // inside a JSON string value (e.g. jdbc:postgresql://host/db URLs) is preserved. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… Snowflake converters
…nk connectors Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Maps Debezium postgres/mysql/sqlserver/oracle connectors to the corresponding *_cdc inputs (JDBC source stays sql_select). MongoDB has no RPCN equivalent and stays unmapped. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ine; oracle dsn + DRY tidy - Stop consuming topic.prefix and database.server.name in all four Debezium mappers (postgres/mysql/sqlserver/oracle) so they fall through to the Unmapped() sweep and appear as inline YAML TODO comments, matching the JDBC convention. - Oracle: annotate connection_string with a TODO when service name is empty. - MySQL: remove spurious "TODO: " prefix from database.server.id warn message. - SQL Server + Oracle: replace inlined table-emit logic with debeziumTables() (same source key table.include.list, same CSV split, same field name). - Regenerate debezium_postgres and debezium_mysql goldens. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…t, insertfield, replacefield, regexrouter)
Gap 1: TimestampConverter now reads the 'format' property and translates
Joda/SimpleDateFormat patterns to Go layouts via jodaToGoLayout(); falls back
to RFC3339 + TODO warn only when 'format' is absent.
Gap 3: castMethod maps int8/16/32/64 to .int64() instead of .number() so
integer columns remain integers.
Gap 7: Cast with a bare type in 'spec' (no colon) is now treated as a
whole-value cast: emits 'root = this.<method>()'.
Gap 5: InsertField now handles topic.field (@kafka_topic), timestamp.field
(kafka_timestamp_ms metadata), partition.field, and offset.field in addition
to static.field/static.value; multiple fields emit one assignment line each.
Gap 6: ReplaceField now supports 'blacklist' as a legacy alias for 'exclude'
and 'include'/'whitelist' projection (root = {} + root.<f> = this.<f>).
Gap 18: ReplaceField passthrough stub comment now accurately names which
properties were looked for instead of hard-coding "include/whitelist semantics".
Gap 17: RegexRouter changed from metadata("kafka_topic").re_replace_all(...) to
@kafka_topic.re_replace_all(...) shorthand; smt_chain golden regenerated.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…Matches/HasHeaderKey predicates Adds smt_filter.go with full support for Kafka Connect's Filter SMT and the predicates framework. The mapper resolves the named predicate type, builds the corresponding Bloblang conditional-delete expression, and consumes all predicates.* keys so they do not surface as unmapped-field warnings. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…DC; recognize ExtractNewRecordState Gap 2: consumeDebeziumCommon now marks value.converter/key.converter (and their sub-keys) consumed so mapConverters() skips emitting schema_registry_decode — *_cdc inputs read structured rows directly from WAL/binlog, not Avro/JSON bytes. Gap 9: new smt_debezium.go registers ExtractNewRecordState as a no-op SMT with an informational warning instead of the generic "unsupported — map manually" stub. debezium_postgres golden input updated to include Avro value.converter to exercise and lock in the fix. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Add injectUserInfo helper that inserts user[:password]@ after the first "://" in the DSN. driverAndDSN now reads and consumes connection.user and connection.password so they no longer surface as unmapped-field warnings. A secret-hygiene TODO comment is appended to the DSN when a password is inlined, matching the Debezium pattern. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…view) Replace double-# TODO concatenation with a single clean semicolon-separated sentence; add special-character caveat on injectUserInfo; pin comment shape in tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…er, compression, time partitioner
- objectFormatExtension: fall back to Aiven format.output.type
(json/jsonl/csv/parquet/avro) when format.class is absent; consume both keys
- mapBatching: take countKeys []string (first-present wins, all consumed); S3
and GCS pass [flush.size, file.max.records] so Aiven file.max.records maps to
batching.count
- file.compression.type: append .gz/.snappy/.zst to the object path extension
plus an inline TODO to add a matching compress processor (connector layer
cannot inject one)
- TimeBasedPartitioner: translate path.format Joda pattern into a time-bucketed
path prefix via jodaToGoLayoutForPath, reusing jodaToGoLayout, emitting
${! metadata("kafka_timestamp_ms").number().ts_format(...) } interpolations
(verified against the benthos linter); consume partitioner.class,
path.format, partition.duration.ms, locale, timezone; best-effort TODO when
path.format is absent/untranslatable
- update remaining mapBatching callers (bigquery, jdbc) to the slice signature
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…regex topics, JSON-Schema converter Gap 13: emit Bloblang match expression over @kafka_topic for multi-entry snowflake.topic2table.map (table field is InterpolatedString; single entry stays literal). Gap 14: add regexp_topics: true to mirror input so Java regex patterns like orders.* are honoured. Gap 10: register io.confluent.connect.json.JsonSchemaConverter as schemaRegistryConverter (same as Avro/Protobuf). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…for sinks When a sink connector specifies topics.regex instead of a literal topics CSV, sinkInputFromTopics now builds a redpanda input with the regex pattern in the topics list and regexp_topics: true, matching how the mirror connector handles regex-based topic selection. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ms→seconds) (review) Switch jodaToGoLayoutForPath from kafka_timestamp_ms (epoch ms) to kafka_timestamp_unix (epoch seconds). Bloblang ts_format interprets a bare number as epoch SECONDS, so feeding it epoch milliseconds produced wildly wrong years (~55000 instead of ~2024). kafka_timestamp_unix is set by record.Timestamp.Unix() in the franz reader and is genuinely epoch seconds. Add hazard comments at the doc and emission site. Pin the corrected expression in TimeBasedPartitioner tests for both S3 and GCS and assert the old broken form is not emitted. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ractTopic, ByLogicalTableRouter, etc.) Add smt_routing.go with five topic-routing SMT mappers: - org.apache.kafka.connect.transforms.TimestampRouter - io.confluent.connect.transforms.MessageTimestampRouter (with message.timestamp.keys TODO/warn) - io.confluent.connect.transforms.ExtractTopic$Value/$Key/$Header - io.confluent.connect.cloud.transforms.TopicRegexRouter (delegates to regexRouterSMT) - io.debezium.transforms.ByLogicalTableRouter (with key.field.* TODO/warn) All emit `meta kafka_topic = <expr>` via mappingProc. TimestampRouter/MessageTimestampRouter build Bloblang concat from topic.format placeholders using kafka_timestamp_unix + ts_format(). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…HeaderFrom, HeaderToValue)
Adds smt_headers.go with four header-manipulation SMT mappers:
- InsertHeader → meta <name> = "<literal>"
- DropHeaders → meta <name> = deleted() per header
- HeaderFrom$Value/$Key → meta <header> = this.<field> (copy); also root.<field> = deleted() on move
- io.debezium.transforms.HeaderToValue → root.<field> = metadata("<header>") (copy/move inverse)
Quoted meta key helper added for hyphenated header names (Bloblang parses bare x-y as subtraction).
Mismatched fields/headers lengths emit a TODO + ctx.Warn for the common prefix.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…tten, GzipDecompress, Drop, TombstoneHandler, FromXml) + Confluent ReplaceField alias Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ox, scripting, partitions, mongodb, geometry) with tailored guidance Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…rocessor When a non-Filter SMT carries transforms.<alias>.predicate (and optional negate), wrap its processor nodes in a benthos switch processor so the transform applies only to records matching the predicate — matching KC conditional-transform semantics generically. Refactored resolvePredicate → resolvePredicateExpr (shared helper); both filterSMT and the new applyPredicateGating in convert.go call it. consumeAllPredicateKeys is called by applyPredicateGating so predicate keys never surface as unmapped fields. Filter SMT (the only KC built-in that handles predicates internally) is exempt from double-gating. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…loblang paths (review) Add fieldPath(base, name) helper in smt_helpers.go that splits KC field names on "." and %q-quotes any segment that is not a safe Bloblang identifier (hyphens, leading digits, etc.). Move isSafeIdentifier there from smt_headers.go (one canonical location). Apply fieldPath to every this.<field> / root.<field> interpolation across all SMT mappers: smt_cast, smt_extractfield, smt_headers, smt_insertfield, smt_maskfield, smt_replacefield, smt_routing, smt_timestampconverter, smt_valuetokey. Safe-identifier names remain bare, so all 18 goldens are unchanged. New tests in smt_fieldpath_test.go cover the helper unit and representative SMTs (ExtractField, InsertField, Cast, ReplaceField, HeaderFrom, ValueToKey) with a hyphenated field "event-type" verifying assertValidRPCN + quoted form. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Blind per-connector sweep (10 real-world KC configs × 11 connectors = 110,
authored without reference to converter support) found 6 lint-breaking configs
and a set of semantic/noise gaps. All are now fixed; re-running the sweep gives
0 convert errors and 0 lint failures.
Lint-breakers:
- G1: recognize Oracle JDBC URLs → `oracle` driver + `oracle://` DSN (was
driver:"" which fails the sql_* driver enum); thin service & SID forms.
- G2: Debezium *_cdc emits a non-empty `tables` placeholder when capture is
defined by exclude-list / DB-level / regex (empty entry failed the
"at least one table" lint).
Correctness:
- G3: JDBC source no longer emits a spurious schema_registry_decode (sql_select
reads DB rows directly) — consumes value/key.converter like the CDC path.
- G4: JDBC sink translates insert.mode=upsert/update into a dialect-aware
ON CONFLICT / ON DUPLICATE KEY suffix and derives columns/args_mapping from
fields.whitelist (MERGE dialects get a clear manual TODO).
- G5: ExtractNewRecordState warning now enumerates the present row-shaping
options (add.fields/add.headers/delete.handling.mode/drop.tombstones).
- G6: JDBC source mode → ORDER BY <cursor> suffix instead of an empty stub.
- G7: MirrorMaker honors replication.policy.class — DefaultReplicationPolicy
prefixes the target topic with the source alias; Identity keeps the name.
- G8: BigQuery derives a topic-based `table` (topic2TableMap match or
${! @kafka_topic }) instead of an empty stub; warns on upsert/delete.
Coverage / noise:
- G9: S3/GCS handle Daily/Hourly partitioners (time-bucket prefix) and
FieldPartitioner (field=${! this.field }/ prefix); consume parquet.codec,
file.name.*, format.output.fields, timestamp.extractor.
- G10: register ByteArrayConverter as a no-op converter (MM2 default / raw).
- G11: consume topic.prefix / database.server.name (no *_cdc equivalent).
- G12: fold sslmode (PG DSN) and encrypt/trustServerCertificate (SQL Server)
into the connection string; surface Oracle PDB; consume adapter/url/log.mining.
- G13: consume schema.history.internal.* / database.history.* families.
- G14: warn on Snowflake schematization; consume streaming/role override opts.
- G15: warn on MirrorMaker topics.exclude; consume groups/offset-syncs/policy.
Adds gap_fixes_test.go regression tests; regenerates affected golden fixtures.
Includes the per-connector gap analysis doc that motivated the work.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…connectors
Two new supported connectors, with class names/aliases/config keys taken from
current official docs (Apache Iceberg, Tabular, Confluent Replicator, MM2).
Iceberg sink → `iceberg` output (REST catalog):
- Classes: org.apache.iceberg.connect.IcebergSinkConnector (current) and the
legacy alias io.tabular.iceberg.connect.IcebergSinkConnector. No distinct
Confluent/Aiven class (both ship the upstream connector).
- iceberg.catalog.uri/warehouse → catalog.url/warehouse; credential (client:secret)
→ catalog.auth.oauth2; token → catalog.auth.bearer.
- AWS Glue (catalog-impl=GlueCatalog) → the Glue REST endpoint + catalog.auth.aws_sigv4
(service=glue) + a derived schema_evolution.table_location (Glue assigns none).
- Non-REST catalogs (hive/hadoop/nessie/jdbc) → stub URL + warning (RPCN iceberg is
REST-only).
- iceberg.tables → namespace + table (last dot splits); multiple tables / dynamic
routing (dynamic-enabled + route-field) → table interpolation + TODO.
- storage backend (aws_s3/gcp_cloud_storage/azure_blob_storage) inferred from io-impl
and the warehouse URI scheme; bucket/container parsed from the warehouse location.
- evolve-schema-enabled → schema_evolution.enabled; control.commit.interval-ms →
batching.period; upsert-mode-enabled/cdc-field → warning (iceberg appends).
- value.converter left to the converter pass (a sink decodes Kafka bytes); the
redpanda input is synthesized from topics/topics.regex by the engine.
Kafka→Kafka (redpanda→redpanda):
- io.confluent.connect.replicator.ReplicatorSourceConnector → redpanda input+output.
src/dest.kafka.bootstrap.servers → seed_brokers; topic.whitelist/topic.regex →
topics (+regexp_topics); src.consumer.group.id → consumer_group; topic.rename.format
${topic} → ${! @kafka_topic } interpolation; src/dest.kafka.security.protocol →
tls/sasl TODO stubs; topic.blacklist → warning; converters consumed (byte copy, no
schema_registry_decode).
- MM2 MirrorCheckpointConnector / MirrorHeartbeatConnector recognized as operational
(drop stub + informational warning), not data pipelines.
Adds conn_iceberg.go/conn_replicator.go (+ tests), 4 golden fixtures, registers the
iceberg component bundle in the lint harness, and updates the README connector table
(also fixes the stale MirrorMaker kafka_franz→redpanda entry).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…nd-trip
Make the converter handle both sides of serialization explicitly instead of
leaving gaps as TODO comments.
Decode (Kafka record → structured):
- JsonConverter is now a real converter, not a silent no-op. schemas.enable=true
emits `root = this.payload` to unwrap Kafka Connect's {schema,payload} envelope;
schemas.enable=false emits nothing (RPCN auto-parses JSON bytes on structured
access); absent → a warning noting the Connect default is true.
- Avro/Protobuf/JSON-Schema continue to emit schema_registry_decode.
Encode (structured → sink file format) for object-store sinks:
- Component gains an Encode []*yaml.Node slot, appended to the pipeline after
converters + SMTs (runs last, just before the output).
- format.class/format.output.type=Avro → an `avro` (operator: from_json) encode
processor with a schema TODO, replacing the old "add an encode step" comment.
- Parquet → a path TODO to add parquet_encode (it can't lint without a schema,
which the converter can't infer).
- JSON/JSONL/CSV/bytes need no encode (the output serializes structured records).
Registers the avro component bundle in the lint harness; regenerates the
s3_sink_full / gcs_sink_full goldens (now carry the avro encode step); adds
decode_encode_test.go and updates the README converter/serialization tables.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Pure reorganization — no behavior change. The package was a flat list of ~60 files with tests interleaved and no obvious entry point. Group by name prefix so the directory listing clusters by role, and add a map so a newcomer knows where to start and how to debug. Renames (git mv, same package — no import changes): - Pipeline core → engine_*: convert/parse/assemble/render (+ matching tests). - yamlutil.go → yaml.go; smt_helpers.go → helpers_smt.go; field_helpers.go → helpers_field.go (+ matching tests). - linter_test.go → testsupport_test.go (the shared test harness) and moved the gapConvert helper into it alongside assertValidRPCN so all shared test infrastructure lives in one place. Adds doc.go: a package overview that walks the Convert() pipeline, maps every file group to its role, and gives a step-by-step debug flow (golden tests + -update, assertValidRPCN, warnings/TODO markers). The README "Architecture" section gains a repository-layout table and the stale filename references (parse.go/assemble.go/render.go/yamlutil.go/linter_test.go) are corrected. Tests stay colocated with their source (Go requires same-package tests for the unexported API); connectors/converters/SMTs keep their already-consistent conn_*/conv_*/smt_* prefixes. lint 0, all converter + CLI tests green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.