debezium/dbz#285 add csv connector example to debezium-examples by yuanglili · Pull Request #407 · debezium/debezium-examples

yuanglili · 2026-04-23T03:20:28Z

Description

This PR adds a CSV file-based connector example to debezium-examples

It demonstrates the key concepts of building a Debezium connector including:

snapshot/streaming handover,
offset management
schema evolution
value type conversion

Checklist

If the changes include a new example, I added it to the list of examples in the README.md file

Signed-off-by: nonononoonononon <yuangli971@gmail.com>

yuanglili · 2026-04-23T03:28:05Z

@kmos, I created an example based on your debezium-connector-csv and it works as expected.

I'm also thinking about whether it would be helpful to add diagrams showing the relationships between the key components, or a simple UI for performing insert/update/delete operations to make it more intuitive.

Let me know if there's anything that needs to be changed!

kmos · 2026-04-23T08:23:21Z

@yuanglili I’m a big supporter of diagrams and visualization. I totally agree.

Naros

Hi @yuanglili, great work. I've left some inline comments to bring the PR more in line with how most Debezium connectors are structured, for those who may use this as a skeleton for future connectors.

Btw, I noticed the PR uses Debezium 3.0.8, which is very old. Could we move this up to Debezium 3.5 by any chance?

Naros · 2026-05-06T19:49:00Z

+
+    @Override
+    public String getContextName() {
+        return "CSV";


Let's add a Module class like other connectors and replace the hard-coded values here for context and connector to refer to the Module class.

Naros · 2026-05-06T19:51:51Z

+    }
+
+    @Override
+    protected SourceInfoStructMaker<?> getSourceInfoStructMaker(Version version) {


Is there a reason for not delegating CommonConnectorConfig#getSourceInfoStructMaker like other connectors? If there is, we should document the reason.

Naros · 2026-05-06T19:53:04Z

+
+    @Override
+    public String version() {
+        return "1.0";


This can also delegate to Module

Naros · 2026-05-06T19:55:10Z

+ * {@link #doStop} is a no-op;
+ * the coordinator manages its own shutdown.
+ */
+public class CsvConnectorTask extends BaseSourceTask<CsvPartition, CsvOffsetContext> {


I believe we may be missing overrides for getConnectorName and getErrorHandler?

Naros · 2026-05-06T19:55:53Z

+import io.debezium.connector.base.ChangeEventQueue;
+import io.debezium.pipeline.ErrorHandler;
+
+public class CsvErrorHandler extends ErrorHandler {


We probably should include some class-level Java docs.

Naros · 2026-05-06T20:07:21Z

+        Schema keySchema = collectionSchema.keySchema();
+        Schema valueSchema = collectionSchema.getEnvelopeSchema().schema()
+                .field("after").schema();
+
+        ColumnDef keyCol = cols.get(0);
+        Struct keyStruct = new Struct(keySchema)
+                .put(keyCol.name(), convertValue(parts[1], keyCol.type()));
+
+        Struct valueStruct = new Struct(valueSchema);
+        for (int i = 0; i < cols.size(); i++) {
+            ColumnDef col = cols.get(i);
+            String rawValue = (i + 1 < parts.length) ? parts[i + 1] : null;
+            valueStruct.put(col.name(), rawValue == null ? null : convertValue(rawValue, col.type()));
+        }


Could we move this logic into the emitter class? In practice, we often centralize this in the emitter because it's shared logic that you'll need for snapshot and streaming, and you reuse the emitter in both contexts.

Naros · 2026-05-06T20:08:45Z

+ * handover line. The fine-grained snapshot/skip decision is made in
+ * {@link CsvSnapshotChangeEventSource#getSnapshottingTask} via the line-number offset.
+ */
+class CsvSnapshotter implements Snapshotter {


Any reason we could not just use the existing InitialSnapshotter from debezium-connector-common?

Naros · 2026-05-06T20:09:27Z

+
+    @Override
+    public String version() {
+        return "1.0";


Another Module.version() reference opportunity.

Naros · 2026-05-06T20:09:50Z

+    @Override
+    public <T extends DataCollectionId> List<T> getMatchingCollections(Configuration configuration) {
+        return List.of();
+    }


I believe this has since been removed, depending on the version of Debezium this is based.

Naros · 2026-05-06T20:11:24Z

+        org.apache.kafka.connect.data.Schema keySchema = collectionSchema.keySchema();
+        org.apache.kafka.connect.data.Schema valueSchema = collectionSchema.getEnvelopeSchema().schema()
+                .field("after").schema();
+
+        ColumnDef keyCol = cols.get(0);
+        org.apache.kafka.connect.data.Struct keyStruct =
+                new org.apache.kafka.connect.data.Struct(keySchema)
+                        .put(keyCol.name(), CsvSnapshotChangeEventSource.convertValue(parts[1], keyCol.type()));
+
+        org.apache.kafka.connect.data.Struct valueStruct =
+                new org.apache.kafka.connect.data.Struct(valueSchema);
+        for (int i = 0; i < cols.size(); i++) {
+            ColumnDef col = cols.get(i);
+            String rawValue = (i + 1 < parts.length) ? parts[i + 1] : null;
+            valueStruct.put(col.name(),
+                    rawValue == null ? null : CsvSnapshotChangeEventSource.convertValue(rawValue, col.type()));
+        }


This is the shared logic I mentioned that we can move into the emitter.

jpechane · 2026-05-07T05:20:33Z

I wonder if it would be possible to extend Relation to the Debezium Framework section signficantly to make sure that every single class necessary to implement the connector (I mean generic connector) is included and short description is added? That would be great start for a follow-up blogpost explaining the details and future conversion into formal connector developer giude.

yuanglili · 2026-05-07T06:45:14Z

Hi @yuanglili, great work. I've left some inline comments to bring the PR more in line with how most Debezium connectors are structured, for those who may use this as a skeleton for future connectors.

Btw, I noticed the PR uses Debezium 3.0.8, which is very old. Could we move this up to Debezium 3.5 by any chance?

Sure, I'll switch it to Debezium 3.5

yuanglili · 2026-05-07T06:48:11Z

I wonder if it would be possible to extend Relation to the Debezium Framework section signficantly to make sure that every single class necessary to implement the connector (I mean generic connector) is included and short description is added? That would be great start for a follow-up blogpost explaining the details and future conversion into formal connector developer giude.

I totally agree. I’m working on a more detailed version of the section right now. I’ll first address the current review comments and then submit an updated version.

yuanglili added 2 commits April 22, 2026 23:15

debezium/dbz#285 add csv connector example to debezium-examples

ff979ea

Signed-off-by: nonononoonononon <yuangli971@gmail.com>

debezium/dbz#285 add csv connector example to README.md

b1631b9

Signed-off-by: nonononoonononon <yuangli971@gmail.com>

kmos requested review from Naros, jpechane, mfvitale and vjuranek April 23, 2026 08:23

Naros reviewed May 6, 2026

View reviewed changes

This was referenced May 13, 2026

Create official documentation for Debezium Source Connector debezium/dbz#1930

Open

Debezium Source Connector Maven archetype - relational non-historized debezium/dbz#1931

Open

Create blog post to how create a Debezim Source Connector debezium/dbz#1932

Open

Uh oh!

Uh oh!

Conversation

yuanglili commented Apr 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Checklist

Uh oh!

yuanglili commented Apr 23, 2026

Uh oh!

kmos commented Apr 23, 2026

Uh oh!

Naros left a comment

Choose a reason for hiding this comment

Uh oh!

Naros May 6, 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!

Naros May 6, 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!

jpechane commented May 7, 2026

Uh oh!

yuanglili commented May 7, 2026

Uh oh!

yuanglili commented May 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

yuanglili commented Apr 23, 2026 •

edited

Loading

Naros May 6, 2026 •

edited

Loading

Naros May 6, 2026 •

edited

Loading