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

README.md

@@ -15,6 +15,7 @@ For getting started please check the [tutorial example](./tutorial).
 * [Cache Invalidation](./cache-invalidation): How Debezium can be used to invalidate items in the JPA 2nd level cache after external data changes
 * [Camel - pipelines](./camel-component): Building an Apache Camel pipeline that captures **Postgres** database changes
 * [Camel - Kafka Connect](./camel-kafka-connect): How to use the Camel Kafka Connect component with Debezium
+* [CSV Connector](./debezium-connector-csv): Example CDC connector implementation based on CSV files, demonstrating snapshot/streaming handover, offset management, and schema evolution
 * [Cloud Events](./cloudevents): How to use cloud events defined in Json with Debezium
 * [Database Activity Monitoring](./db-activity-monitoring): How to use Debezium for comprehensive database activity logging and analysis
 * [Debezium - End-to-end demo](./end-to-end-demo): End-to-end demo using MySQL as database and Kafka Connect

debezium-connector-csv/.gitignore

@@ -0,0 +1,7 @@
+.idea/
+*.iml
+
+target/
+
+dependency-reduced-pom.xml
+settings.local.json

debezium-connector-csv/README.md

@@ -0,0 +1,182 @@
+# Debezium CSV Connector
+
+A reference implementation of a [Debezium](https://debezium.io) source connector designed to show **how to build a CDC connector** on top of the Debezium 3.x framework (the same architecture used by the PostgreSQL, SQL Server and Oracle connectors).
+
+---
+
+## Overview
+
+The connector treats a single log-style file as a "database". Every line in the file represents a change event. A Java NIO `WatchService` detects when the file is modified and the connector reads newly appended lines.
+
+This is intentionally simple so that the code serves as a guide for the framework concepts rather than a production-grade solution.
+
+---
+
+## File Format
+
+```
+EventType;Id:INT;FirstName:STRING;LastName:STRING;Salary:INT;Active:BOOLEAN
+S|1|Alice|Smith|3000|true
+S|2|Bob|Miller|2000|false
+I|3|Bruce|Masters|4000|false
+U|1|Sarah|Smith|3000|false
+D|3|Bruce|Masters|4000|false
+EventType;Id:INT;FirstName:STRING;LastName:STRING;Salary:INT;Active:BOOLEAN;Birthday:DATE
+U|1|Sarah|Smith|3000|false|1980-12-31
+```
+
+### Schema header lines
+
+A line that starts with `EventType` defines the schema for all subsequent data rows:
+
+```
+EventType;Id:INT;FirstName:STRING;LastName:STRING;Salary:INT;Active:BOOLEAN
+```
+
+- Fields are separated by `;`
+- The format of each field spec is `name` or `name:TYPE`
+- `EventType` is a literal keyword that identifies a header line — it is **not** a data column
+- Supported types: `STRING` (default), `INT`, `LONG`, `FLOAT`, `DOUBLE`, `BOOLEAN`, `DATE` (ISO-8601, e.g. `1980-12-31`)
+- A new schema header anywhere in the file triggers **schema evolution**: all subsequent rows use the new column definitions
+
+### Data rows
+
+```
+<EventType>|<col1>|<col2>|…
+```
+
+| Character | Meaning | Debezium `op` |
+|-----------|---------|---------------|
+| `S` | Snapshot — initial table state | `r` (READ) |
+| `I` | Insert | `c` (CREATE) |
+| `U` | Update (after-image only) | `u` (UPDATE) |
+| `D` | Delete | `d` (DELETE) |
+
+---
+
+## Kafka Topic
+
+Each connector instance monitors one file. The topic name is:
+
+```
+<topic.prefix>.<filename-without-extension>
+```
+
+For example, with `topic.prefix=hr` and `csv.file.path=/data/employees.csv` the topic is `hr.employees`.
+
+---
+
+## Configuration
+
+| Property | Required | Default | Description |
+|----------|----------|---------|-------------|
+| `connector.class` | yes | — | `io.debezium.connector.csv.CsvSourceConnector` |
+| `topic.prefix` | yes | — | Logical server name; used as topic prefix |
+| `csv.file.path` | yes | — | Absolute path to the CSV database file |
+
+All [common Debezium connector properties](https://debezium.io/documentation/reference/stable/connectors/index.html) (poll interval, batch size, etc.) are also supported.
+
+### Minimal example
+
+```properties
+connector.class=io.debezium.connector.csv.CsvSourceConnector
+topic.prefix=hr
+csv.file.path=/data/employees.csv
+```
+
+---
+
+## Key Design Concepts
+
+### Offset — line number
+
+The connector tracks progress using a **line number** (zero-based index of the next line to read), stored in the Connect offset storage as `{"line": N}`.
+
+This makes it trivial to resume after a restart and to hand off from the snapshot phase to the streaming phase.
+
+Relevant classes: `CsvOffsetContext`, `CsvOffsetLoader`
+
+### Snapshot vs. Streaming
+
+```
+CsvSnapshotChangeEventSource                CsvStreamingChangeEventSource
+────────────────────────────                ─────────────────────────────
+Read file from line 0                       Start from handover line
+Process header → register schema            Watch for file modifications (WatchService)
+Emit READ for every S row                   Emit CREATE / UPDATE / DELETE for I/U/D rows
+Stop at first non-S, non-header line   ──►  Continue from that line
+```
+
+The `ChangeEventSourceCoordinator` orchestrates the transition: it runs the snapshot source first, saves the resulting offset, then starts the streaming source from that offset.
+
+Relevant classes: `CsvSnapshotChangeEventSource`, `CsvStreamingChangeEventSource`, `CsvChangeEventSourceFactory`
+
+### Schema evolution
+
+A new schema header line mid-file causes `CsvSchema.register()` to rebuild the Kafka Connect key/value/envelope schemas. Because all value fields are declared `OPTIONAL`, existing consumers that have not yet received the new field will treat it as `null`.
+
+Relevant class: `CsvSchema`
+
+### Type conversion
+
+`ColumnType` enumerates the supported column types. `CsvSnapshotChangeEventSource.convertValue()` performs the string-to-Java conversion (e.g. `"3000"` → `Integer`, `"1980-12-31"` → days-since-epoch `int`).
+
+Relevant classes: `ColumnType`, `ColumnDef`, `CsvSchema.toKafkaSchema()`
+
+### Source info
+
+Every change event includes a `source` block with Debezium metadata plus two CSV-specific fields:
+- `file` — name of the database file
+- `line` — zero-based line number of the record
+
+Relevant classes: `CsvSourceInfo`, `CsvSourceInfoStructMaker`
+
+---
+
+## Running
+
+**1. Build the connector JAR:**
+
+```bash
+mvn package
+```
+
+Requires Java 21+ and Maven 3.
+
+**2. Start Kafka, Kafka Connect, and the Kafka UI:**
+
+```bash
+docker compose up -d
+```
+
+**3. Register the connector:**
+
+```bash
+curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
+  http://localhost:8083/connectors/ -d @register-csv.json
+```
+
+The connector will snapshot the initial `S` rows from `src/main/resources/employees.csv` and then stream any new lines appended to the file. Browse events at [http://localhost:8080](http://localhost:8080).
+
+---
+
+## Sample File
+
+A sample database file is included at `src/main/resources/employees.csv`.
+
+---
+
+## Relation to the Debezium Framework
+
+| Framework concept | CSV implementation |
+|-------------------|--------------------|
+| `DataCollectionId` | `CsvId` (table name = filename without extension) |
+| `Partition` / `Partition.Provider` | `CsvPartition` / `CsvProvider` (keyed on `topic.prefix`) |
+| `OffsetContext` / `OffsetContext.Loader` | `CsvOffsetContext` / `CsvOffsetLoader` (stores `{"line": N}`) |
+| `DatabaseSchema` | `CsvSchema` (lazily built from schema header lines) |
+| `SnapshotChangeEventSource` | `CsvSnapshotChangeEventSource` |
+| `StreamingChangeEventSource` | `CsvStreamingChangeEventSource` |
+| `ChangeEventSourceFactory` | `CsvChangeEventSourceFactory` |
+| `ChangeRecordEmitter` | `CsvChangeRecordEmitter` |
+| `Snapshotter` | `CsvSnapshotter` (initial snapshot only) |
+| `SourceInfoStructMaker` | `CsvSourceInfoStructMaker` |

debezium-connector-csv/docker-compose.yml

@@ -0,0 +1,42 @@
+services:
+  kafka:
+    image: apache/kafka:3.8.0
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: broker,controller
+      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+
+  connect:
+    image: quay.io/debezium/connect:3.0
+    ports:
+      - "8083:8083"
+    environment:
+      - BOOTSTRAP_SERVERS=kafka:9092
+      - GROUP_ID=1
+      - CONFIG_STORAGE_TOPIC=connect_configs
+      - OFFSET_STORAGE_TOPIC=connect_offsets
+      - STATUS_STORAGE_TOPIC=connect_statuses
+    volumes:
+      - ./target/csv-connector:/kafka/connect/csv-connector
+      - ./src/main/resources:/data
+    depends_on:
+      - kafka
+
+  kafka-ui:
+    image: provectuslabs/kafka-ui:latest
+    ports:
+      - "8080:8080"
+    environment:
+      - KAFKA_CLUSTERS_0_NAME=local
+      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
+    depends_on:
+      - kafka

debezium-connector-csv/pom.xml

@@ -0,0 +1,118 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>io.debezium.connector.csv</groupId>
+    <artifactId>debezium-connector-csv</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <name>debezium-connector-csv</name>
+    <url>https://www.debezium.io</url>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <maven.compiler.release>21</maven.compiler.release>
+        <version.debezium>3.0.8.Final</version.debezium>
+    </properties>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.junit</groupId>
+                <artifactId>junit-bom</artifactId>
+                <version>5.11.0</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>io.debezium</groupId>
+                <artifactId>debezium-bom</artifactId>
+                <version>${version.debezium}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <pluginManagement>
+            <plugins>
+                <plugin>
+                    <artifactId>maven-clean-plugin</artifactId>
+                    <version>3.4.0</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-resources-plugin</artifactId>
+                    <version>3.3.1</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-compiler-plugin</artifactId>
+                    <version>3.13.0</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-surefire-plugin</artifactId>
+                    <version>3.3.0</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-jar-plugin</artifactId>
+                    <version>3.4.2</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-install-plugin</artifactId>
+                    <version>3.1.2</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-deploy-plugin</artifactId>
+                    <version>3.1.2</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-site-plugin</artifactId>
+                    <version>3.12.1</version>
+                </plugin>
+                <plugin>
+                    <artifactId>maven-project-info-reports-plugin</artifactId>
+                    <version>3.6.1</version>
+                </plugin>
+            </plugins>
+        </pluginManagement>
+
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>3.5.0</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

debezium-connector-csv/register-csv.json

@@ -0,0 +1,9 @@
+{
+    "name": "csv-connector",
+    "config": {
+        "connector.class": "io.debezium.connector.csv.CsvSourceConnector",
+        "tasks.max": "1",
+        "topic.prefix": "hr",
+        "csv.file.path": "/data/employees.csv"
+    }
+}