diff --git a/README.md b/README.md index 251a6e73b7..4bc31f3917 100644 --- a/README.md +++ b/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 diff --git a/debezium-connector-csv/.gitignore b/debezium-connector-csv/.gitignore new file mode 100644 index 0000000000..492739f727 --- /dev/null +++ b/debezium-connector-csv/.gitignore @@ -0,0 +1,7 @@ +.idea/ +*.iml + +target/ + +dependency-reduced-pom.xml +settings.local.json \ No newline at end of file diff --git a/debezium-connector-csv/README.md b/debezium-connector-csv/README.md new file mode 100644 index 0000000000..2e766cf229 --- /dev/null +++ b/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 + +``` +|||… +``` + +| 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: + +``` +. +``` + +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` | diff --git a/debezium-connector-csv/docker-compose.yml b/debezium-connector-csv/docker-compose.yml new file mode 100644 index 0000000000..7bdf37d7ad --- /dev/null +++ b/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 diff --git a/debezium-connector-csv/pom.xml b/debezium-connector-csv/pom.xml new file mode 100644 index 0000000000..fbaa5dcae9 --- /dev/null +++ b/debezium-connector-csv/pom.xml @@ -0,0 +1,118 @@ + + + 4.0.0 + + io.debezium.connector.csv + debezium-connector-csv + 1.0-SNAPSHOT + + debezium-connector-csv + https://www.debezium.io + + + UTF-8 + 21 + 3.0.8.Final + + + + + + org.junit + junit-bom + 5.11.0 + pom + import + + + io.debezium + debezium-bom + ${version.debezium} + pom + import + + + + + + + io.debezium + debezium-core + + + org.apache.kafka + connect-api + provided + + + org.slf4j + slf4j-api + provided + + + org.junit.jupiter + junit-jupiter-api + test + + + + + + + + maven-clean-plugin + 3.4.0 + + + maven-resources-plugin + 3.3.1 + + + maven-compiler-plugin + 3.13.0 + + + maven-surefire-plugin + 3.3.0 + + + maven-jar-plugin + 3.4.2 + + + maven-install-plugin + 3.1.2 + + + maven-deploy-plugin + 3.1.2 + + + maven-site-plugin + 3.12.1 + + + maven-project-info-reports-plugin + 3.6.1 + + + + + + + org.apache.maven.plugins + maven-shade-plugin + 3.5.0 + + + package + + shade + + + + + + + \ No newline at end of file diff --git a/debezium-connector-csv/register-csv.json b/debezium-connector-csv/register-csv.json new file mode 100644 index 0000000000..0fcb9ca8c3 --- /dev/null +++ b/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" + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/ColumnDef.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/ColumnDef.java new file mode 100644 index 0000000000..66d175a011 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/ColumnDef.java @@ -0,0 +1,36 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +/** + * Definition of a single column parsed from a CSV schema header. + * + * @param name the column name + * @param type the column type + */ +record ColumnDef(String name, ColumnType type) { + + /** + * Parses a column spec of the form {@code name} or {@code name:TYPE}. + * Unknown type tokens are treated as {@link ColumnType#STRING}. + */ + static ColumnDef parse(String spec) { + int colon = spec.indexOf(':'); + if (colon < 0) { + return new ColumnDef(spec.trim(), ColumnType.STRING); + } + String name = spec.substring(0, colon).trim(); + String typePart = spec.substring(colon + 1).trim().toUpperCase(); + ColumnType type; + try { + type = ColumnType.valueOf(typePart); + } + catch (IllegalArgumentException e) { + type = ColumnType.STRING; + } + return new ColumnDef(name, type); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/ColumnType.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/ColumnType.java new file mode 100644 index 0000000000..b467587541 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/ColumnType.java @@ -0,0 +1,24 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +/** + * Supported column types in a CSV schema header. + * + *

Example header line: + * {@code EventType;Id:INT;FirstName:STRING;Salary:INT;Active:BOOLEAN;Birthday:DATE} + * + *

If no type is specified the column defaults to {@link #STRING}. + */ +enum ColumnType { + STRING, + INT, + LONG, + FLOAT, + DOUBLE, + BOOLEAN, + DATE +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeEventSourceFactory.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeEventSourceFactory.java new file mode 100644 index 0000000000..61cbbe65a1 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeEventSourceFactory.java @@ -0,0 +1,50 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.pipeline.EventDispatcher; +import io.debezium.pipeline.notification.NotificationService; +import io.debezium.pipeline.source.spi.ChangeEventSourceFactory; +import io.debezium.pipeline.source.spi.SnapshotChangeEventSource; +import io.debezium.pipeline.source.spi.SnapshotProgressListener; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; + +/** + * Creates the snapshot and streaming change event sources for the CSV connector. + * + *

Called by {@link io.debezium.pipeline.ChangeEventSourceCoordinator} once per connector start. + * Both sources share the same {@link CsvConnectorConfig}, {@link CsvId}, {@link CsvSchema} + * and {@link EventDispatcher} instances that are wired together in {@link CsvConnectorTask}. + */ +class CsvChangeEventSourceFactory implements ChangeEventSourceFactory { + + private final CsvConnectorConfig config; + private final CsvId csvId; + private final CsvSchema schema; + private final EventDispatcher dispatcher; + + CsvChangeEventSourceFactory(CsvConnectorConfig config, + CsvId csvId, + CsvSchema schema, + EventDispatcher dispatcher) { + this.config = config; + this.csvId = csvId; + this.schema = schema; + this.dispatcher = dispatcher; + } + + @Override + public SnapshotChangeEventSource getSnapshotChangeEventSource( + SnapshotProgressListener progressListener, + NotificationService notificationService) { + return new CsvSnapshotChangeEventSource(config, csvId, schema, dispatcher, progressListener); + } + + @Override + public StreamingChangeEventSource getStreamingChangeEventSource() { + return new CsvStreamingChangeEventSource(config, csvId, schema, dispatcher); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeEventSourceMetricsFactory.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeEventSourceMetricsFactory.java new file mode 100644 index 0000000000..e6ddc4717f --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeEventSourceMetricsFactory.java @@ -0,0 +1,39 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.connector.base.ChangeEventQueueMetrics; +import io.debezium.connector.common.CdcSourceTaskContext; +import io.debezium.pipeline.metrics.DefaultChangeEventSourceMetricsFactory; +import io.debezium.pipeline.metrics.SnapshotChangeEventSourceMetrics; +import io.debezium.pipeline.metrics.StreamingChangeEventSourceMetrics; +import io.debezium.pipeline.metrics.spi.ChangeEventSourceMetricsFactory; +import io.debezium.pipeline.source.spi.EventMetadataProvider; + +/** + * Provides standard Debezium JMX metrics for snapshot and streaming phases. + * + *

Delegates to {@link DefaultChangeEventSourceMetricsFactory}, which exposes + * generic metrics (records captured, events per second, etc.) without requiring + * any connector-specific instrumentation. + */ +class CsvChangeEventSourceMetricsFactory implements ChangeEventSourceMetricsFactory { + + private final DefaultChangeEventSourceMetricsFactory delegate = + new DefaultChangeEventSourceMetricsFactory<>(); + + @Override + public SnapshotChangeEventSourceMetrics getSnapshotMetrics( + T context, ChangeEventQueueMetrics queueMetrics, EventMetadataProvider metadataProvider) { + return delegate.getSnapshotMetrics(context, queueMetrics, metadataProvider); + } + + @Override + public StreamingChangeEventSourceMetrics getStreamingMetrics( + T context, ChangeEventQueueMetrics queueMetrics, EventMetadataProvider metadataProvider) { + return delegate.getStreamingMetrics(context, queueMetrics, metadataProvider); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeRecordEmitter.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeRecordEmitter.java new file mode 100644 index 0000000000..9060307f68 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvChangeRecordEmitter.java @@ -0,0 +1,89 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.data.Envelope; +import io.debezium.pipeline.spi.ChangeRecordEmitter; +import io.debezium.pipeline.spi.OffsetContext; +import io.debezium.schema.DataCollectionSchema; +import org.apache.kafka.connect.data.Struct; + +import java.time.Instant; + +/** + * Wraps a pre-parsed CSV row and emits it as a Debezium change record. + * + *

Called by {@link io.debezium.pipeline.EventDispatcher} after schema lookup. + * Builds the full Kafka Connect envelope struct (op, before, after, source, ts_ms) + * and hands it to the dispatcher's internal receiver. + * + *

UPDATE events carry no before-image; {@code before} is always {@code null}. + */ +class CsvChangeRecordEmitter implements ChangeRecordEmitter { + + private final CsvPartition partition; + private final CsvOffsetContext offsetContext; + private final Envelope.Operation operation; + private final Struct keyStruct; + private final Struct valueStruct; + + CsvChangeRecordEmitter(CsvPartition partition, + CsvOffsetContext offsetContext, + Envelope.Operation operation, + Struct keyStruct, + Struct valueStruct) { + this.partition = partition; + this.offsetContext = offsetContext; + this.operation = operation; + this.keyStruct = keyStruct; + this.valueStruct = valueStruct; + } + + /** + * Constructs the envelope struct for the current operation and dispatches it. + * + * @param schema resolved collection schema; provides the {@link Envelope} builder + * @param receiver dispatcher-provided receiver that enqueues the final SourceRecord + * @throws InterruptedException if the receiver's queue is interrupted + * + *

NOTE: the value passed to {@code receiver.changeRecord()} must be the complete + * envelope struct, not the raw value struct — a mismatch causes + * {@code DataException: Mismatching schema} in the JSON converter. + */ + @Override + public void emitChangeRecords(DataCollectionSchema schema, Receiver receiver) + throws InterruptedException { + Envelope envelope = schema.getEnvelopeSchema(); + Struct source = offsetContext.getSourceInfoStruct(); + Instant now = Instant.now(); + + // Envelope API: read/create(after, source, ts) · update(before, after, source, ts) · delete(before, source, ts) + Struct envelopeStruct = switch (operation) { + case READ -> envelope.read(valueStruct, source, now); + case CREATE -> envelope.create(valueStruct, source, now); + case UPDATE -> envelope.update(null, valueStruct, source, now); + case DELETE -> envelope.delete(valueStruct, source, now); + default -> throw new IllegalStateException("Unexpected operation: " + operation); + }; + + receiver.changeRecord(partition, schema, operation, keyStruct, envelopeStruct, offsetContext, null); + } + + @Override + public CsvPartition getPartition() { + return partition; + } + + @Override + public OffsetContext getOffset() { + return offsetContext; + } + + @Override + public Envelope.Operation getOperation() { + return operation; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvCollectionSchema.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvCollectionSchema.java new file mode 100644 index 0000000000..5d5be8a869 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvCollectionSchema.java @@ -0,0 +1,49 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.data.Envelope; +import io.debezium.schema.DataCollectionSchema; +import org.apache.kafka.connect.data.Schema; + +/** + * Kafka Connect schema for the single CSV "table". + * + *

The key schema is a struct containing only the primary-key column + * (the first column defined in the CSV schema header). + * + * The value schema is wrapped in a Debezium {@link Envelope} that adds + * the standard {@code op}, {@code ts_ms}, {@code before}, {@code after} and {@code source} fields. + * + *

Instances are created and cached by {@link CsvSchema#register}. + */ +class CsvCollectionSchema implements DataCollectionSchema { + + private final CsvId id; + private final Schema keySchema; + private final Envelope envelope; + + CsvCollectionSchema(CsvId id, Schema keySchema, Envelope envelope) { + this.id = id; + this.keySchema = keySchema; + this.envelope = envelope; + } + + @Override + public CsvId id() { + return id; + } + + @Override + public Schema keySchema() { + return keySchema; + } + + @Override + public Envelope getEnvelopeSchema() { + return envelope; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvConnectorConfig.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvConnectorConfig.java new file mode 100644 index 0000000000..a27fcf6588 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvConnectorConfig.java @@ -0,0 +1,106 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.config.CommonConnectorConfig; +import io.debezium.config.Configuration; +import io.debezium.config.EnumeratedValue; +import io.debezium.config.Field; +import io.debezium.connector.SourceInfoStructMaker; +import org.apache.kafka.common.config.ConfigDef; + +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Optional; + +/** + * Connector configuration for the CSV connector. + * + *

Extends {@link CommonConnectorConfig} with one connector-specific field: + * {@link #FILE_PATH}. All other settings (topic prefix, poll interval, batch size) + * are inherited. The file's base name (without extension) becomes the Kafka topic suffix. + * + *

Minimum required properties: {@code connector.class}, {@code topic.prefix}, + * {@code csv.file.path}. + */ +public class CsvConnectorConfig extends CommonConnectorConfig { + + /** + * Snapshot strategy enum. Only {@code INITIAL} is supported: snapshot on first start, + * stream from the stored line-number offset on subsequent starts. + */ + public enum SnapshotMode implements EnumeratedValue { + INITIAL("initial"); + + private final String value; + + SnapshotMode(String value) { + this.value = value; + } + + @Override + public String getValue() { + return value; + } + } + + /** + * Absolute path to the CSV database file (e.g. {@code /data/employees.csv}). + * The filename without extension becomes the table name and Kafka topic suffix. + */ + public static final Field FILE_PATH = Field.create("csv.file.path") + .withDisplayName("CSV file path") + .withType(ConfigDef.Type.STRING) + .withWidth(ConfigDef.Width.LONG) + .withImportance(ConfigDef.Importance.HIGH) + .withDescription("Absolute path to the CSV database file, e.g. /data/employees.csv") + .required(); + + static final Field.Set ALL_FIELDS = Field.setOf(FILE_PATH); + + public CsvConnectorConfig(Configuration config, int defaultSnapshotFetchSize) { + super(config, defaultSnapshotFetchSize); + } + + /** Returns the Kafka Connect {@link ConfigDef} used to validate configuration at deployment. */ + public static ConfigDef configDef() { + ConfigDef config = new ConfigDef(); + Field.group(config, "CSV", FILE_PATH); + return config; + } + + public Path getFilePath() { + return Paths.get(getConfig().getString(FILE_PATH)); + } + + @Override + public String getContextName() { + return "CSV"; + } + + @Override + public String getConnectorName() { + return "csv"; + } + + @Override + public EnumeratedValue getSnapshotMode() { + return SnapshotMode.INITIAL; + } + + @Override + public Optional getSnapshotLockingMode() { + return Optional.empty(); + } + + @Override + protected SourceInfoStructMaker getSourceInfoStructMaker(Version version) { + // init() must be called here — CommonConnectorConfig never calls it, but AbstractSourceInfo.schema() needs it. + CsvSourceInfoStructMaker maker = new CsvSourceInfoStructMaker(); + maker.init(getConnectorName(), "1.0", this); + return maker; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvConnectorTask.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvConnectorTask.java new file mode 100644 index 0000000000..0a30281955 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvConnectorTask.java @@ -0,0 +1,194 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.config.Configuration; +import io.debezium.config.Field; +import io.debezium.connector.base.ChangeEventQueue; +import io.debezium.connector.common.BaseSourceTask; +import io.debezium.connector.common.CdcSourceTaskContext; +import io.debezium.document.DocumentReader; +import io.debezium.pipeline.ChangeEventSourceCoordinator; +import io.debezium.pipeline.DataChangeEvent; +import io.debezium.pipeline.ErrorHandler; +import io.debezium.pipeline.EventDispatcher; +import io.debezium.pipeline.metrics.spi.ChangeEventSourceMetricsFactory; +import io.debezium.pipeline.notification.NotificationService; +import io.debezium.pipeline.signal.SignalProcessor; +import io.debezium.pipeline.source.spi.ChangeEventSourceFactory; +import io.debezium.pipeline.spi.Offsets; +import io.debezium.schema.DataCollectionFilters; +import io.debezium.schema.SchemaFactory; +import io.debezium.snapshot.SnapshotterService; +import io.debezium.snapshot.lock.NoLockingSupport; +import io.debezium.spi.topic.TopicNamingStrategy; +import org.apache.kafka.connect.source.SourceRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Map; + +/** + * Kafka Connect {@link org.apache.kafka.connect.source.SourceTask} for the CSV connector. + * + *

{@link #start} wires all Debezium infrastructure (config, schema, dispatcher, metrics, coordinator) + * and starts the coordinator, which runs snapshot and streaming phases on background threads. + * {@link #doPoll} drains the internal + * {@link ChangeEventQueue} on each Connect poll cycle. + * {@link #doStop} is a no-op; + * the coordinator manages its own shutdown. + */ +public class CsvConnectorTask extends BaseSourceTask { + + private static final Logger LOGGER = LoggerFactory.getLogger(CsvConnectorTask.class); + + /** Held as a field so that {@link #doPoll()} can drain it. */ + private volatile ChangeEventQueue changeEventQueue; + private volatile ErrorHandler errorHandler; + + @Override + protected ChangeEventSourceCoordinator start(Configuration configuration) { + + // 1. config + CsvConnectorConfig connectorConfig = new CsvConnectorConfig(configuration, 0); + + // 2. derive table name from filename (e.g. employees.csv → "employees") + String fileName = connectorConfig.getFilePath().getFileName().toString(); + String tableName = fileName.contains(".") + ? fileName.substring(0, fileName.lastIndexOf('.')) + : fileName; + CsvId csvId = new CsvId(tableName); + + // 3. restore stored offsets; on first start getTheOnlyOffset() returns null — seed a + // zero-offset context so snapshot logic sees a valid (not null) offsetContext + String serverName = connectorConfig.getLogicalName(); + Offsets previousOffsets = getPreviousOffsets( + new CsvProvider(serverName), + new CsvOffsetLoader(connectorConfig)); + + if (previousOffsets.getTheOnlyOffset() == null) { + CsvPartition partition = new CsvPartition(serverName); + CsvOffsetContext offsetContext = new CsvOffsetContext(new CsvSourceInfo(connectorConfig)); + previousOffsets = Offsets.of(partition, offsetContext); + } + + // 4. task context (needed before queue so it can supply a logging-context) + CdcSourceTaskContext taskContext = new CdcSourceTaskContext( + connectorConfig, Map.of(), () -> List.of(csvId)); + + // 5. change-event queue buffers records between producer threads and poll() + changeEventQueue = new ChangeEventQueue.Builder() + .pollInterval(connectorConfig.getPollInterval()) + .maxBatchSize(connectorConfig.getMaxBatchSize()) + .maxQueueSize(connectorConfig.getMaxQueueSize()) + .loggingContextSupplier(() -> taskContext.configureLoggingContext("csv")) + .buffering() + .build(); + + // 6. error handler routes unrecoverable errors back through the queue + errorHandler = new CsvErrorHandler(connectorConfig, changeEventQueue, errorHandler); + + // 7. JMX metrics + ChangeEventSourceMetricsFactory metricsFactory = + new CsvChangeEventSourceMetricsFactory(); + + // 8. topic naming strategy: . + TopicNamingStrategy topicNamingStrategy = + connectorConfig.getTopicNamingStrategy(CsvConnectorConfig.TOPIC_NAMING_STRATEGY); + + // 9. schema holder — populated lazily from the CSV header line + CsvSchema schema = new CsvSchema(); + + // 10. include-all filter (single-file connector) + DataCollectionFilters.DataCollectionFilter dataCollectionFilter = + new CsvDataCollectionFilter(); + + // 11. signal processor (Debezium signal channels) + SignalProcessor signalProcessor = new SignalProcessor<>( + CsvSourceConnector.class, connectorConfig, Map.of(), + getAvailableSignalChannels(), + DocumentReader.defaultReader(), + previousOffsets); + + // 12. event metadata provider + CsvMetadataProvider eventMetadataProvider = new CsvMetadataProvider(); + + // 13. dispatcher routes records through schema/filter into the change-event queue + EventDispatcher eventDispatcher = new EventDispatcher<>( + connectorConfig, + topicNamingStrategy, + schema, + changeEventQueue, + dataCollectionFilter, + DataChangeEvent::new, + eventMetadataProvider, + connectorConfig.schemaNameAdjuster()); + + // 14. notification service (heartbeats, schema-change notifications) + NotificationService notificationService = + new NotificationService<>( + getNotificationChannels(), + connectorConfig, + SchemaFactory.get(), + eventDispatcher::enqueueNotification); + + // 15. snapshotter: initial snapshot on first start, stream thereafter + SnapshotterService snapshotterService = new SnapshotterService( + new CsvSnapshotter(), + new CsvSnapshotQuery(), + new NoLockingSupport()); + + // 16. event source factory — must be created AFTER dispatcher (captures a reference to it) + ChangeEventSourceFactory csvEventSourceFactory = + new CsvChangeEventSourceFactory(connectorConfig, csvId, schema, eventDispatcher); + + // 17. coordinator orchestrates the snapshot → streaming transition + ChangeEventSourceCoordinator coordinator = + new ChangeEventSourceCoordinator<>( + previousOffsets, + errorHandler, + CsvSourceConnector.class, + connectorConfig, + csvEventSourceFactory, + metricsFactory, + eventDispatcher, + schema, + signalProcessor, + notificationService, + snapshotterService); + + coordinator.start(taskContext, changeEventQueue, eventMetadataProvider); + + return coordinator; + } + + /** + * Drains the {@link ChangeEventQueue} and returns the batch to the Connect framework. + * Blocks up to {@code poll.interval.ms} before returning an empty list. + */ + @Override + protected List doPoll() throws InterruptedException { + return changeEventQueue.poll().stream() + .map(DataChangeEvent::getRecord) + .toList(); + } + + /** No-op — the coordinator manages its own shutdown. */ + @Override + protected void doStop() { + } + + @Override + protected Iterable getAllConfigurationFields() { + return CsvConnectorConfig.ALL_FIELDS; + } + + @Override + public String version() { + return "1.0"; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvDataCollectionFilter.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvDataCollectionFilter.java new file mode 100644 index 0000000000..f93240c6d9 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvDataCollectionFilter.java @@ -0,0 +1,20 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; +import io.debezium.schema.DataCollectionFilters; + +/** + * Determines which data collections to include when producing change events. + * + * The CSV connector tracks a single file, so all collections are always included. + */ +class CsvDataCollectionFilter implements DataCollectionFilters.DataCollectionFilter { + + @Override + public boolean isIncluded(CsvId csvId) { + return true; + } +} \ No newline at end of file diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvErrorHandler.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvErrorHandler.java new file mode 100644 index 0000000000..140aecae67 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvErrorHandler.java @@ -0,0 +1,16 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.config.CommonConnectorConfig; +import io.debezium.connector.base.ChangeEventQueue; +import io.debezium.pipeline.ErrorHandler; + +public class CsvErrorHandler extends ErrorHandler { + public CsvErrorHandler(CommonConnectorConfig connectorConfig, ChangeEventQueue queue, ErrorHandler replacedErrorHandler) { + super(CsvSourceConnector.class, connectorConfig, queue, replacedErrorHandler); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvId.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvId.java new file mode 100644 index 0000000000..9e26978589 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvId.java @@ -0,0 +1,53 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.spi.schema.DataCollectionId; + +import java.util.List; + +/** + * Identifies the single "table" tracked by a CSV connector instance. + * + *

The identifier is the base name of the database file (filename without extension), + * e.g. {@code employees} for {@code /data/employees.csv}. + * This value is also used as the table segment in the Kafka topic name: + * {@code .}. + */ +public class CsvId implements DataCollectionId { + + private final String tableName; + + public CsvId(String tableName) { + this.tableName = tableName; + } + + /** Used by the framework for topic naming and offset storage. */ + @Override + public String identifier() { + return tableName; + } + + @Override + public List parts() { + return List.of(tableName); + } + + @Override + public List databaseParts() { + return List.of(); + } + + @Override + public List schemaParts() { + return List.of(tableName); + } + + @Override + public String toString() { + return tableName; + } +} \ No newline at end of file diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvMetadataProvider.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvMetadataProvider.java new file mode 100644 index 0000000000..3070a0c6a7 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvMetadataProvider.java @@ -0,0 +1,39 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.pipeline.source.spi.EventMetadataProvider; +import io.debezium.pipeline.spi.OffsetContext; +import io.debezium.spi.schema.DataCollectionId; +import org.apache.kafka.connect.data.Struct; + +import java.time.Instant; +import java.util.Map; + +/** + * Provides event-level metadata used by the framework for logging and monitoring. + */ +class CsvMetadataProvider implements EventMetadataProvider { + + @Override + public Instant getEventTimestamp(DataCollectionId dataCollectionId, OffsetContext offsetContext, Object key, Struct value) { + return Instant.now(); + } + + @Override + public Map getEventSourcePosition(DataCollectionId dataCollectionId, OffsetContext offsetContext, Object key, Struct value) { + if (offsetContext instanceof CsvOffsetContext ctx) { + return Map.of(CsvOffsetContext.LINE_OFFSET_KEY, String.valueOf(ctx.getLineNumber())); + } + return Map.of(); + } + + /** CSV has no transactions; return empty string. */ + @Override + public String getTransactionId(DataCollectionId dataCollectionId, OffsetContext offsetContext, Object key, Struct value) { + return ""; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvOffsetContext.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvOffsetContext.java new file mode 100644 index 0000000000..bf7519d7c1 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvOffsetContext.java @@ -0,0 +1,75 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.pipeline.CommonOffsetContext; +import io.debezium.pipeline.txmetadata.TransactionContext; +import io.debezium.spi.schema.DataCollectionId; +import org.apache.kafka.connect.data.Schema; + +import java.time.Instant; +import java.util.Map; + +/** + * Tracks the current CSV read position. + * + *

The offset is stored as {@code {"line": N}}, where {@code N} is the next line to read. + * This is used for restart recovery and snapshot-to-streaming handoff. + */ +public class CsvOffsetContext extends CommonOffsetContext { + + static final String LINE_OFFSET_KEY = "line"; + + /** Zero-based index of the next line to read. */ + private long lineNumber; + + public CsvOffsetContext(CsvSourceInfo sourceInfo) { + super(sourceInfo); + } + + public long getLineNumber() { + return lineNumber; + } + + public void setLineNumber(long lineNumber) { + this.lineNumber = lineNumber; + } + + /** + * Serialises the current position into the map that is persisted by + * the Kafka Connect offset storage. + */ + @Override + public Map getOffset() { + return Map.of(LINE_OFFSET_KEY, lineNumber); + } + + @Override + public Schema getSourceInfoSchema() { + return sourceInfo.schema(); + } + + /** + * Called by the framework before each event is enqueued. + * Updates the source info so that the line number is reflected in the + * {@code source} block of the outgoing message. + */ + @Override + public void event(DataCollectionId dataCollectionId, Instant instant) { + sourceInfo.setLineNumber(lineNumber); + } + + /** CSV has no transactions; return an empty context. */ + @Override + public TransactionContext getTransactionContext() { + return new TransactionContext(); + } + + /** Returns the source info as a Struct for embedding in envelope records. */ + org.apache.kafka.connect.data.Struct getSourceInfoStruct() { + return sourceInfo.struct(); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvOffsetLoader.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvOffsetLoader.java new file mode 100644 index 0000000000..c666c2adff --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvOffsetLoader.java @@ -0,0 +1,39 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.pipeline.spi.OffsetContext; + +import java.util.Map; + +/** + * Restores a {@link CsvOffsetContext} from Kafka Connect's persisted offset map. + * + *

The stored map format is {@code {"line": N}}. On first start the map is + * {@code null}/empty, producing a zero-offset context that triggers a fresh snapshot. + * On subsequent starts streaming resumes from the persisted line number. + */ +public class CsvOffsetLoader implements OffsetContext.Loader { + + private final CsvConnectorConfig csvConnectorConfig; + + public CsvOffsetLoader(CsvConnectorConfig csvConnectorConfig) { + this.csvConnectorConfig = csvConnectorConfig; + } + + @Override + public CsvOffsetContext load(Map map) { + CsvOffsetContext ctx = new CsvOffsetContext(new CsvSourceInfo(csvConnectorConfig)); + if (map != null) { + Object lineValue = map.get(CsvOffsetContext.LINE_OFFSET_KEY); + if (lineValue != null) { + ctx.setLineNumber(Long.parseLong(lineValue.toString())); + } + } + // lineNumber stays 0 when map is null/empty → triggers snapshot on first start + return ctx; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvPartition.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvPartition.java new file mode 100644 index 0000000000..993ccb51fd --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvPartition.java @@ -0,0 +1,37 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.pipeline.spi.Partition; + +import java.util.Map; + +/** + * Identifies the offset partition for a CSV connector instance. + * + *

Each connector instance tracks a single file. The partition key is the logical server name ({@code topic.prefix}), + * so two connector instances targeting different files under different prefixes keep independent offsets. + */ +public class CsvPartition implements Partition { + + private static final String SERVER_KEY = "server"; + + private final String serverName; + + public CsvPartition(String serverName) { + this.serverName = serverName; + } + + /** + * Returns the Kafka Connect source partition map. + * This map is used as the key when storing/loading offsets via the + * Connect framework's offset storage. + */ + @Override + public Map getSourcePartition() { + return Map.of(SERVER_KEY, serverName); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvProvider.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvProvider.java new file mode 100644 index 0000000000..5a07bbb8cc --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvProvider.java @@ -0,0 +1,31 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.pipeline.spi.Partition; + +import java.util.Collections; +import java.util.Set; + +/** + * Supplies the set of partitions that this connector instance manages. + * + *

A CSV connector tracks exactly one file, so there is always a single + * partition keyed on the logical server name ({@code topic.prefix}). + */ +public class CsvProvider implements Partition.Provider { + + private final String serverName; + + public CsvProvider(String serverName) { + this.serverName = serverName; + } + + @Override + public Set getPartitions() { + return Collections.singleton(new CsvPartition(serverName)); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSchema.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSchema.java new file mode 100644 index 0000000000..58194034d4 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSchema.java @@ -0,0 +1,110 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.config.CommonConnectorConfig; +import io.debezium.data.Envelope; +import io.debezium.schema.DataCollectionSchema; +import io.debezium.schema.DatabaseSchema; +import org.apache.kafka.connect.data.Date; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.SchemaBuilder; + +import java.util.List; + +/** + * Holds the active Kafka Connect schema (key, value, envelope) for the CSV table. + * + *

Schema is derived from the CSV header line and rebuilt whenever a new header is encountered (schema evolution). + * Callers invoke {@link #register} to update it. + * Schema names follow the pattern {@code ..Key/.Value/.Envelope}. + */ +class CsvSchema implements DatabaseSchema { + + private CsvCollectionSchema cachedSchema; + + /** + * Builds and caches key/value/envelope schemas from the given column definitions. + * Called on the initial schema header and again on each schema-evolution header. + * + * @param id table identifier used for schema naming + * @param cols ordered column definitions; first column is treated as the primary key + * @param cfg connector config; provides logical name and the source-info struct maker + */ + void register(CsvId id, List cols, CommonConnectorConfig cfg) { + String prefix = cfg.getLogicalName() + "." + id.identifier(); + + // Key schema: single-field struct containing only the primary key (first column) + ColumnDef keyCol = cols.get(0); + Schema keySchema = SchemaBuilder.struct() + .name(prefix + ".Key") + .field(keyCol.name(), toKafkaSchema(keyCol.type())) + .build(); + + // Value schema must be optional: Envelope uses it directly as the "before"/"after" field + // schema, and "before" is null for READ/CREATE events — non-optional would cause serialisation failure. + SchemaBuilder valueBuilder = SchemaBuilder.struct().name(prefix + ".Value"); + for (ColumnDef col : cols) { + valueBuilder.field(col.name(), toKafkaSchema(col.type())); + } + Schema valueSchema = valueBuilder.optional().build(); + + Schema sourceSchema = cfg.getSourceInfoStructMaker().schema(); + + // Envelope adds op / ts_ms / source / before / after fields around the value schema + Envelope envelope = Envelope.defineSchema() + .withName(prefix + ".Envelope") + .withRecord(valueSchema) + .withSource(sourceSchema) + .build(); + + cachedSchema = new CsvCollectionSchema(id, keySchema, envelope); + } + + /** Returns the most recently registered schema, or {@code null} if none registered yet. */ + @Override + public DataCollectionSchema schemaFor(CsvId csvId) { + return cachedSchema; + } + + /** Typed variant of {@link #schemaFor}. */ + CsvCollectionSchema getCurrentSchema() { + return cachedSchema; + } + + /** {@code true} once the first schema header has been processed. */ + @Override + public boolean tableInformationComplete() { + return cachedSchema != null; + } + + /** No schema history topic needed for CSV. */ + @Override + public boolean isHistorized() { + return false; + } + + @Override + public void close() { + // nothing to close + } + + /** + * Maps a {@link ColumnType} to the corresponding Kafka Connect {@link Schema}. + * All schemas are optional to support schema evolution (adding columns without breaking consumers). + */ + static Schema toKafkaSchema(ColumnType type) { + return switch (type) { + case INT -> Schema.OPTIONAL_INT32_SCHEMA; + case LONG -> Schema.OPTIONAL_INT64_SCHEMA; + case FLOAT -> Schema.OPTIONAL_FLOAT32_SCHEMA; + case DOUBLE -> Schema.OPTIONAL_FLOAT64_SCHEMA; + case BOOLEAN -> Schema.OPTIONAL_BOOLEAN_SCHEMA; + case DATE -> Date.builder().optional().build(); + default -> Schema.OPTIONAL_STRING_SCHEMA; + }; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotChangeEventSource.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotChangeEventSource.java new file mode 100644 index 0000000000..fca2816af3 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotChangeEventSource.java @@ -0,0 +1,239 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.data.Envelope; +import io.debezium.pipeline.EventDispatcher; +import io.debezium.pipeline.source.SnapshottingTask; +import io.debezium.pipeline.source.spi.ChangeEventSource; +import io.debezium.pipeline.source.spi.SnapshotChangeEventSource; +import io.debezium.pipeline.source.spi.SnapshotProgressListener; +import io.debezium.pipeline.spi.SnapshotResult; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.Struct; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.file.Files; +import java.time.LocalDate; +import java.time.ZoneOffset; +import java.util.Date; +import java.util.List; +import java.util.Map; + +/** + * Reads snapshot rows ({@code S|…}) from the CSV file and emits READ events. + * + *

Processes lines from the beginning: registers schema on each header line, emits a READ event per S-row, + * and stops at the first non-S data line (handing off to streaming) or at end of file. + * + * Updates {@link CsvOffsetContext} to the handover line so streaming resumes at the correct + * position. Skipped entirely when an offset already exists. + */ +class CsvSnapshotChangeEventSource implements SnapshotChangeEventSource { + + private static final Logger LOGGER = LoggerFactory.getLogger(CsvSnapshotChangeEventSource.class); + + /** Separator used between columns in data rows. */ + static final String DATA_SEPARATOR = "\\|"; + + /** Prefix that identifies a schema-header line. */ + static final String HEADER_PREFIX = "EventType"; + + /** Separator used between column specs in a schema header line. */ + static final String HEADER_SEPARATOR = ";"; + + private final CsvConnectorConfig config; + private final CsvId csvId; + private final CsvSchema schema; + private final EventDispatcher dispatcher; + private final SnapshotProgressListener progressListener; + + CsvSnapshotChangeEventSource(CsvConnectorConfig config, + CsvId csvId, + CsvSchema schema, + EventDispatcher dispatcher, + SnapshotProgressListener progressListener) { + this.config = config; + this.csvId = csvId; + this.schema = schema; + this.dispatcher = dispatcher; + this.progressListener = progressListener; + } + + @Override + public SnapshottingTask getSnapshottingTask(CsvPartition partition, CsvOffsetContext offsetContext) { + boolean shouldSnapshot = offsetContext.getLineNumber() == 0; + LOGGER.info("Snapshot decision for partition {}: shouldSnapshot={}, currentLine={}", + partition, shouldSnapshot, offsetContext.getLineNumber()); + return new SnapshottingTask(shouldSnapshot, false, List.of(), Map.of(), false); + } + + @Override + public SnapshottingTask getBlockingSnapshottingTask(CsvPartition partition, CsvOffsetContext offsetContext, + io.debezium.pipeline.signal.actions.snapshotting.SnapshotConfiguration config) { + // CSV snapshot is sequential — blocking and non-blocking modes are equivalent. + return getSnapshottingTask(partition, offsetContext); + } + + /** + * Reads all lines from the start of the file, emitting READ events for S-rows. + * Stops and returns {@link SnapshotResult#completed} at the first non-S line or end of file. + * + * @param task carries {@code shouldSkipSnapshot}; skipped when a prior offset exists + * @throws InterruptedException if the change-event queue is interrupted during dispatch + */ + @Override + public SnapshotResult execute(ChangeEventSource.ChangeEventSourceContext context, + CsvPartition partition, + CsvOffsetContext offsetContext, + SnapshottingTask task) + throws InterruptedException { + + if (task.shouldSkipSnapshot()) { + LOGGER.info("Skipping snapshot – resuming streaming from line {}", offsetContext.getLineNumber()); + return SnapshotResult.skipped(offsetContext); + } + + LOGGER.info("Starting snapshot of {}", config.getFilePath()); + + List lines; + try { + lines = Files.readAllLines(config.getFilePath()); + } + catch (IOException e) { + throw new UncheckedIOException("Cannot read CSV file " + config.getFilePath(), e); + } + + List currentSchema = null; + int snapshotCount = 0; + + for (int i = 0; i < lines.size(); i++) { + if (!context.isRunning()) { + return SnapshotResult.aborted(); + } + + String line = lines.get(i); + + if (isHeader(line)) { + // Schema header line: register the column definitions + currentSchema = parseHeader(line); + schema.register(csvId, currentSchema, config); + LOGGER.debug("Registered schema at line {}: {}", i, currentSchema); + } + else if (line.startsWith("S|")) { + if (currentSchema == null) { + throw new IllegalStateException( + "Data row at line " + i + " precedes any schema header in " + config.getFilePath()); + } + dispatch(Envelope.Operation.READ, line, i, currentSchema, partition, offsetContext); + offsetContext.setLineNumber(i + 1); + snapshotCount++; + } + else { + // first non-S, non-header line: hand off to streaming + offsetContext.setLineNumber(i); + LOGGER.info("Snapshot complete: {} records read, handing over to streaming at line {}", snapshotCount, i); + return SnapshotResult.completed(offsetContext); + } + } + + // end of file — all lines were S-rows or headers + offsetContext.setLineNumber(lines.size()); + LOGGER.info("Snapshot complete: {} records read (end of file at line {})", snapshotCount, lines.size()); + return SnapshotResult.completed(offsetContext); + } + + /** Returns {@code true} if the line is a schema header. */ + static boolean isHeader(String line) { + return line.startsWith(HEADER_PREFIX); + } + + /** + * Parses a schema-header line into an ordered list of {@link ColumnDef}s. + * The leading {@code EventType} field is skipped; remaining fields are {@code name} or {@code name:TYPE}. + * + * @param line a header line starting with {@code EventType;…} + * @return ordered column definitions; first entry is the primary key + */ + static List parseHeader(String line) { + String[] parts = line.split(HEADER_SEPARATOR); + // skip parts[0] ("EventType") + List cols = new java.util.ArrayList<>(parts.length - 1); + for (int i = 1; i < parts.length; i++) { + cols.add(ColumnDef.parse(parts[i])); + } + return List.copyOf(cols); + } + + /** + * Parses a pipe-delimited data row and dispatches it as a change event. + * + * @param op CDC operation (READ, CREATE, UPDATE, DELETE) + * @param line raw CSV line; {@code parts[0]} is the event-type char, {@code parts[1..n]} are values + * @param lineNumber zero-based line index in the file + * @param cols active column definitions; {@code cols[0]} is the primary key + * @param partition source partition + * @param offsetContext current offset; updated via {@code event()} before dispatch + */ + void dispatch(Envelope.Operation op, String line, int lineNumber, + List cols, CsvPartition partition, + CsvOffsetContext offsetContext) throws InterruptedException { + + String[] parts = line.split(DATA_SEPARATOR, -1); // parts[0]=event-type, parts[1..n]=values + + CsvCollectionSchema collectionSchema = schema.getCurrentSchema(); + 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())); + } + + offsetContext.event(csvId, null); + + dispatcher.dispatchDataChangeEvent(partition, csvId, + new CsvChangeRecordEmitter(partition, offsetContext, op, keyStruct, valueStruct)); + } + + /** + * Converts a raw CSV string to the Java type required by the Kafka Connect schema. + * + * @param raw raw string from the CSV; {@code null} or empty returns {@code null} + * @param type target column type + * @return typed value, or {@code null} if raw is empty/null + * + *

NOTE: DATE must be {@link java.util.Date} (logical representation), not the INT32 + * wire value — the converter (e.g. JsonConverter) handles the encoding. + */ + static Object convertValue(String raw, ColumnType type) { + if (raw == null || raw.isEmpty()) { + return null; + } + return switch (type) { + case INT -> Integer.parseInt(raw); + case LONG -> Long.parseLong(raw); + case FLOAT -> Float.parseFloat(raw); + case DOUBLE -> Double.parseDouble(raw); + case BOOLEAN -> Boolean.parseBoolean(raw); + case DATE -> { + LocalDate date = LocalDate.parse(raw); + yield Date.from(date.atStartOfDay(ZoneOffset.UTC).toInstant()); + } + default -> raw; + }; + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotQuery.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotQuery.java new file mode 100644 index 0000000000..d2a34468cc --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotQuery.java @@ -0,0 +1,35 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.snapshot.spi.SnapshotQuery; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +/** + * No-op {@link SnapshotQuery} for the CSV connector. + * CSV data is read directly from the file; SQL queries are not applicable. + */ +class CsvSnapshotQuery implements SnapshotQuery { + + @Override + public String name() { + return "csv"; + } + + @Override + public Optional snapshotQuery(String s, List list) { + // CSV data is read directly from the file, not via a query + return Optional.empty(); + } + + @Override + public void configure(Map properties) { + // no configuration needed + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotter.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotter.java new file mode 100644 index 0000000000..efa00b7cd2 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSnapshotter.java @@ -0,0 +1,61 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.spi.snapshot.Snapshotter; + +import java.util.Map; + +/** + * Initial-mode snapshot strategy for the CSV connector. + * + *

Triggers a snapshot when no offset has been stored yet, then streams from the + * handover line. The fine-grained snapshot/skip decision is made in + * {@link CsvSnapshotChangeEventSource#getSnapshottingTask} via the line-number offset. + */ +class CsvSnapshotter implements Snapshotter { + + @Override + public String name() { + return "initial"; + } + + /** + * @param offsetExists {@code false} on first start (no stored offset) + * @param snapshotInProgress {@code true} if a previous snapshot was interrupted + */ + @Override + public boolean shouldSnapshotData(boolean offsetExists, boolean snapshotInProgress) { + return !offsetExists || snapshotInProgress; + } + + /** Schema is derived from the CSV header line, not from a separate schema snapshot. */ + @Override + public boolean shouldSnapshotSchema(boolean offsetExists, boolean snapshotInProgress) { + return false; + } + + /** Always stream after snapshot. */ + @Override + public boolean shouldStream() { + return true; + } + + @Override + public boolean shouldSnapshotOnSchemaError() { + return false; + } + + @Override + public boolean shouldSnapshotOnDataError() { + return false; + } + + @Override + public void configure(Map properties) { + // no configuration needed + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceConnector.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceConnector.java new file mode 100644 index 0000000000..6f67057551 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceConnector.java @@ -0,0 +1,85 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.config.Configuration; +import io.debezium.connector.common.BaseSourceConnector; +import io.debezium.spi.schema.DataCollectionId; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigValue; +import org.apache.kafka.connect.connector.Task; + +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.List; +import java.util.Map; + +/** + * Top-level Kafka Connect connector for the CSV connector. + * + *

Always runs exactly one {@link CsvConnectorTask} — one connector instance monitors + * one file. {@link #validateAllFields} checks that the configured file path exists at + * deployment time, providing an early failure instead of a cryptic runtime error. + */ +public class CsvSourceConnector extends BaseSourceConnector { + + private Map properties; + + @Override + public void start(Map properties) { + this.properties = Map.copyOf(properties); + } + + @Override + public Class taskClass() { + return CsvConnectorTask.class; + } + + /** + * Returns a single-element list — this connector always uses exactly one task. + * + * @param maxTasks ignored; single-file connectors do not parallelize across tasks + */ + @Override + public List> taskConfigs(int maxTasks) { + return List.of(Map.copyOf(properties)); + } + + @Override + public void stop() { + // Kafka Connect manages task lifecycle; nothing to do here. + } + + @Override + public ConfigDef config() { + return CsvConnectorConfig.configDef(); + } + + @Override + public String version() { + return "1.0"; + } + + /** Validates that the configured CSV file path exists on the local filesystem. */ + @Override + protected Map validateAllFields(Configuration configuration) { + Map results = new java.util.LinkedHashMap<>(); + + String rawPath = configuration.getString(CsvConnectorConfig.FILE_PATH); + if (rawPath != null && !rawPath.isBlank() && !Files.exists(Paths.get(rawPath))) { + ConfigValue cv = new ConfigValue(CsvConnectorConfig.FILE_PATH.name()); + cv.addErrorMessage("File does not exist: " + rawPath); + results.put(CsvConnectorConfig.FILE_PATH.name(), cv); + } + + return results; + } + + @Override + public List getMatchingCollections(Configuration configuration) { + return List.of(); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceInfo.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceInfo.java new file mode 100644 index 0000000000..67a5fed3de --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceInfo.java @@ -0,0 +1,59 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.config.CommonConnectorConfig; +import io.debezium.connector.common.BaseSourceInfo; + +import java.time.Instant; + +/** + * Carries the {@code source} metadata block that is included in every change event. + * + *

For CSV the relevant source information is the name of the file being + * monitored and the line number of the record that triggered the event. + * Both values are populated by {@link CsvOffsetContext#event} before each event + * is emitted, and exposed via {@link CsvSourceInfoStructMaker}. + */ +public class CsvSourceInfo extends BaseSourceInfo { + + private final CommonConnectorConfig config; + private String fileName = ""; + private long lineNumber; + + public CsvSourceInfo(CommonConnectorConfig config) { + super(config); + this.config = config; + } + + public void setFileName(String fileName) { + this.fileName = fileName; + } + + public String getFileName() { + return fileName; + } + + public void setLineNumber(long lineNumber) { + this.lineNumber = lineNumber; + } + + public long getLineNumber() { + return lineNumber; + } + + /** Returns the current wall-clock time as the event timestamp. */ + @Override + protected Instant timestamp() { + return Instant.now(); + } + + /** Returns the logical server name ({@code topic.prefix}) as the database name. */ + @Override + protected String database() { + return config.getLogicalName(); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceInfoStructMaker.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceInfoStructMaker.java new file mode 100644 index 0000000000..7d3841d185 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvSourceInfoStructMaker.java @@ -0,0 +1,49 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.config.CommonConnectorConfig; +import io.debezium.connector.AbstractSourceInfoStructMaker; +import org.apache.kafka.connect.data.Schema; +import org.apache.kafka.connect.data.Struct; + +/** + * Builds the Kafka Connect schema and struct for the {@code source} field that + * appears in every change event envelope. + * + *

In addition to the common Debezium fields (version, connector, name, ts_ms, + * snapshot), the CSV source block includes: + *

    + *
  • {@code file} – the name of the CSV database file
    • + *
  • {@code line} – the zero-based line number of the record in that file
    • + *
+ */ +class CsvSourceInfoStructMaker extends AbstractSourceInfoStructMaker { + + private Schema schema; + + @Override + public void init(String connector, String version, CommonConnectorConfig config) { + super.init(connector, version, config); + schema = commonSchemaBuilder() + .name("io.debezium.connector.csv.Source") + .field("file", Schema.STRING_SCHEMA) + .field("line", Schema.INT64_SCHEMA) + .build(); + } + + @Override + public Schema schema() { + return schema; + } + + @Override + public Struct struct(CsvSourceInfo info) { + return commonStruct(info) + .put("file", info.getFileName()) + .put("line", info.getLineNumber()); + } +} diff --git a/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvStreamingChangeEventSource.java b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvStreamingChangeEventSource.java new file mode 100644 index 0000000000..8fe4715496 --- /dev/null +++ b/debezium-connector-csv/src/main/java/io/debezium/connector/csv/CsvStreamingChangeEventSource.java @@ -0,0 +1,213 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.csv; + +import io.debezium.data.Envelope; +import io.debezium.pipeline.EventDispatcher; +import io.debezium.pipeline.source.spi.ChangeEventSource; +import io.debezium.pipeline.source.spi.StreamingChangeEventSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.file.FileSystems; +import java.nio.file.Files; +import java.nio.file.StandardWatchEventKinds; +import java.nio.file.WatchKey; +import java.nio.file.WatchService; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static io.debezium.connector.csv.CsvSnapshotChangeEventSource.DATA_SEPARATOR; +import static io.debezium.connector.csv.CsvSnapshotChangeEventSource.isHeader; +import static io.debezium.connector.csv.CsvSnapshotChangeEventSource.parseHeader; + +/** + * Streams ongoing changes by tailing the CSV file from the snapshot handover line. + * + *

Uses Java NIO {@link WatchService} to detect file modifications, then reads all + * newly appended lines. Event type mapping: {@code I}→CREATE, {@code U}→UPDATE + * (no before-image), {@code D}→DELETE. A new schema-header line at any position + * re-registers the schema via {@link CsvSchema#register} and applies it to subsequent rows. + * + *

The line-number offset is incremented after every processed line (header or data) + * so that streaming resumes at the exact position on restart. + */ +class CsvStreamingChangeEventSource implements StreamingChangeEventSource { + + private static final Logger LOGGER = LoggerFactory.getLogger(CsvStreamingChangeEventSource.class); + + private final CsvConnectorConfig config; + private final CsvId csvId; + private final CsvSchema schema; + private final EventDispatcher dispatcher; + + CsvStreamingChangeEventSource(CsvConnectorConfig config, + CsvId csvId, + CsvSchema schema, + EventDispatcher dispatcher) { + this.config = config; + this.csvId = csvId; + this.schema = schema; + this.dispatcher = dispatcher; + } + + /** + * Main streaming loop. Registers a {@link WatchService} on the file's parent directory, + * reads newly appended lines on each modification event, and dispatches change events. + * Polls every 1 s to avoid missing rapid consecutive writes. + * + * @throws InterruptedException if the change-event queue is interrupted + */ + @Override + public void execute(ChangeEventSource.ChangeEventSourceContext context, + CsvPartition partition, + CsvOffsetContext offsetContext) throws InterruptedException { + + try (WatchService watchService = FileSystems.getDefault().newWatchService()) { + config.getFilePath().getParent() + .register(watchService, StandardWatchEventKinds.ENTRY_MODIFY); + + // schema may already be populated by snapshot; initialise from cache if not (e.g. all S-rows) + List currentSchema = deriveCurrentSchema(); + long lineNum = offsetContext.getLineNumber(); + + LOGGER.info("Starting streaming from line {} in {}", lineNum, config.getFilePath()); + + while (context.isRunning()) { + // Read any lines that have been appended since last iteration + List lines; + try { + lines = Files.readAllLines(config.getFilePath()); + } + catch (IOException e) { + throw new UncheckedIOException("Cannot read CSV file " + config.getFilePath(), e); + } + + while (lineNum < lines.size()) { + String line = lines.get((int) lineNum); + + if (isHeader(line)) { + // Schema evolution: new column definitions + currentSchema = parseHeader(line); + schema.register(csvId, currentSchema, config); + LOGGER.info("Schema evolved at line {}: {}", lineNum, currentSchema); + } + else { + // Data row: I / U / D + Envelope.Operation op = parseOperation(line); + if (op != null && currentSchema != null) { + dispatchDataLine(op, line, (int) lineNum, currentSchema, partition, offsetContext); + } + else if (currentSchema == null) { + LOGGER.warn("Skipping line {} – no schema registered yet", lineNum); + } + } + + lineNum++; + offsetContext.setLineNumber(lineNum); + } + + WatchKey key = watchService.poll(1, TimeUnit.SECONDS); + if (key != null) { + key.pollEvents(); // drain events to prevent WatchService queue overflow + key.reset(); + } + } + } + catch (IOException e) { + throw new UncheckedIOException("WatchService error for " + config.getFilePath(), e); + } + } + + /** + * Maps the leading character of a data line to a Debezium {@link Envelope.Operation}. + * Returns {@code null} for unknown or {@code S} lines; callers should skip {@code null}. + */ + private static Envelope.Operation parseOperation(String line) { + if (line.isEmpty()) { + return null; + } + return switch (line.charAt(0)) { + case 'I' -> Envelope.Operation.CREATE; + case 'U' -> Envelope.Operation.UPDATE; + case 'D' -> Envelope.Operation.DELETE; + default -> null; // includes 'S' which should not appear in streaming + }; + } + + /** + * Reconstructs the active {@link ColumnDef} list from the cached schema. + * Used when streaming starts without having seen a header line (snapshot already registered it). + * Returns {@code null} if no schema has been registered yet. + */ + private List deriveCurrentSchema() { + CsvCollectionSchema cached = schema.getCurrentSchema(); + if (cached == null) { + return null; + } + // rebuild ColumnDef list from the value schema's "after" fields + return cached.getEnvelopeSchema().schema() + .field("after").schema().fields().stream() + .map(f -> new ColumnDef(f.name(), inferColumnType(f.schema()))) + .toList(); + } + + /** Reverse-maps a Kafka Connect Schema back to a {@link ColumnType}. */ + private static ColumnType inferColumnType(org.apache.kafka.connect.data.Schema s) { + if (org.apache.kafka.connect.data.Date.LOGICAL_NAME.equals(s.name())) { + return ColumnType.DATE; + } + return switch (s.type()) { + case INT32 -> ColumnType.INT; + case INT64 -> ColumnType.LONG; + case FLOAT32 -> ColumnType.FLOAT; + case FLOAT64 -> ColumnType.DOUBLE; + case BOOLEAN -> ColumnType.BOOLEAN; + default -> ColumnType.STRING; + }; + } + + /** + * Parses a streaming data row and dispatches it as a change event. + * + * @param op CDC operation (CREATE, UPDATE, DELETE) + * @param line raw CSV line; {@code parts[0]} is the event-type char + * @param lineNumber zero-based line index (for logging only) + * @param cols active column definitions; {@code cols[0]} is the primary key + */ + private void dispatchDataLine(Envelope.Operation op, String line, int lineNumber, + List cols, CsvPartition partition, + CsvOffsetContext offsetContext) throws InterruptedException { + + String[] parts = line.split(DATA_SEPARATOR, -1); // parts[0]=event-type, parts[1..n]=values + + CsvCollectionSchema collectionSchema = schema.getCurrentSchema(); + 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())); + } + + offsetContext.event(csvId, null); + + dispatcher.dispatchDataChangeEvent(partition, csvId, + new CsvChangeRecordEmitter(partition, offsetContext, op, keyStruct, valueStruct)); + } +} diff --git a/debezium-connector-csv/src/main/resources/employees.csv b/debezium-connector-csv/src/main/resources/employees.csv new file mode 100644 index 0000000000..5094594922 --- /dev/null +++ b/debezium-connector-csv/src/main/resources/employees.csv @@ -0,0 +1,6 @@ +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