Spooldir Source Connector - AutoMQ Documentation

Overview

The Spooldir Source Connector discovers files in a local or shared directory accessible to the Connect Worker, converts their contents into Kafka messages, and writes the messages to a specified topic. It is suitable for ingesting batch-exported files, application logs, or file-based data exchanges into Kafka. Each task scans the input directory, filters files by name, reads records from them, and moves, deletes, or retains the source files after successful or failed processing according to the configured policy. The plugin package provides seven Source Connector implementations. SpoolDirCsvSourceConnector reads CSV and can use explicit schemas or generate schemas from sample files. SpoolDirJsonSourceConnector reads consecutive JSON root values and outputs strongly typed records using explicit or generated schemas. SpoolDirSchemaLessJsonSourceConnector outputs consecutive JSON root values as strings without schemas. SpoolDirLineDelimitedSourceConnector outputs each line as a string. SpoolDirBinaryFileSourceConnector outputs an entire file as one byte-array message. SpoolDirAvroSourceConnector reads Avro container files and uses the schema embedded in each file. SpoolDirELFSourceConnector parses Extended Log Format files and dynamically constructs record schemas. The seven implementations share file discovery, file filtering, processing markers, and cleanup policies, but their record structures, timestamps, batch boundaries, and offset recovery behavior after a restart are not identical. CSV is the primary path used in this document’s Quick Start. When selecting a connector.class for another format, use only the configurations actually supported by that implementation and do not carry over CSV-specific configurations.

Prerequisites

If you extract and deploy the official Confluent Hub 2.0.71 ZIP directly, provide com.google.guava:guava:31.1-jre in the same plugin directory, or use an installation method that resolves this runtime dependency into the plugin classloader. The ZIP does not include Guava; without it, plugin discovery or task startup can fail. Restart the Connect Worker after adding the dependency.
The Connect Worker must have read and write access to input.path and error.path. When using the MOVE or MOVEBYDATE cleanup policy, it must also be able to access and write to finished.path. Create these directories before starting the Connector.
When using schema generation for CSV or strongly typed JSON, place at least one regular file matching input.file.pattern at the top level of input.path before starting the Connector, and ensure that the sampled files generate consistent schemas.

License

Uses Apache License 2.0.

Quick Start

Prepare a Connect Cluster, Kafka, the topic that will receive the data, and input, finished, and error directories that the Worker can read and write. Confirm file system access permissions. For preparation and management, see Manage connectors.

connector.class=com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector
topic=<topic-name>
input.path=<input-directory>
finished.path=<finished-directory>
error.path=<error-directory>
input.file.pattern=^.*\.csv$
schema.generation.enabled=true
schema.generation.key.fields=id
csv.first.row.as.header=true

Replace the placeholders with actual resources. Before starting the Connector, place at least one file ending in .csv at the top level of <input-directory>. The first row must be a header and must contain an id column. The Connector generates a schema with string fields from the matching file, writes id to the message key, and moves the file to <finished-directory> after successful processing under the default MOVE policy. Files that cannot be processed are moved to <error-directory>.

Configuration

Output and File Discovery

`topic`

The Kafka topic that receives records from files.

Type: string
Default: None
Importance: High
Valid Values / Notes: Must be a valid topic name that the Connector can write to. An empty string is not rejected by the plugin’s configuration validation, but it cannot serve as a usable destination.
Required: Yes

`input.path`

The input directory containing files to process.

Type: string
Default: None
Importance: High
Valid Values / Notes: The directory must already exist, and the Connect process must have read and write permissions. By default, only direct child files are scanned.
Required: Yes

`input.file.pattern`

The Java regular expression used to select input file names.

Type: string
Default: None
Importance: High
Valid Values / Notes: Matches the complete file name, not the complete path. It cannot be empty and must be a valid regular expression. A single match operation that takes longer than 100 milliseconds is treated as no match.
Required: Yes

`file.minimum.age.ms`

The minimum time in milliseconds since a file was last modified before it can be discovered.

Type: long
Default: 0
Importance: Low
Valid Values / Notes: Must be greater than or equal to 0. This condition is checked only during file discovery; it does not lock the file or verify again that the file is no longer being written.

`input.path.walk.recursively`

Whether to scan descendant files under input.path recursively.

Type: boolean
Default: false
Importance: Low
Valid Values / Notes: When set to true, input.file.pattern is still applied to each file’s name.

`files.sort.attributes`

The sort attributes applied before candidate files enter the task queue.

Type: list
Default: NameAsc
Importance: Low
Valid Values / Notes: Each item can be NameAsc, NameDesc, LengthAsc, LengthDesc, LastModifiedAsc, or LastModifiedDesc; comparisons are combined in list order. Sorting constrains only the local file queue of a single task and does not provide global ordering across tasks.

`empty.poll.wait.ms`

The time in milliseconds that a task waits after consecutive empty reads.

Type: long
Default: 500
Importance: Low
Valid Values / Notes: The range is 1 through Long.MAX_VALUE. It controls sleeping after empty reads and does not guarantee a fixed file discovery interval.

File Lifecycle and Error Handling

`cleanup.policy`

How successfully processed files are cleaned up.

Type: string
Default: MOVE
Importance: Medium
Valid Values / Notes: Valid options are NONE, DELETE, MOVE, and MOVEBYDATE. Both MOVE and MOVEBYDATE require a writable finished.path. Files that fail processing are always moved to error.path when possible.

`finished.path`

The destination directory for successfully processed files.

Type: string
Default: Empty string
Importance: High
Valid Values / Notes: Required when using MOVE or MOVEBYDATE. The directory must exist and be writable before the task starts.

`error.path`

The destination directory for files that fail processing.

Type: string
Default: None
Importance: High
Valid Values / Notes: The directory must already exist and be writable. Even when halt.on.error=false, a failed file is first moved to this directory when possible.
Required: Yes

`halt.on.error`

Whether a file processing failure causes the task to fail.

Type: boolean
Default: true
Importance: High
Valid Values / Notes: true throws an exception after cleaning up the failed file. false attempts to move the file to error.path and then continues selecting other files. Error handling is performed at whole-file granularity, not per record.

`processing.file.extension`

The suffix used for processing marker files.

Type: string
Default: .PROCESSING
Importance: Low
Valid Values / Notes: Appended to the input file name. The value must contain a period and a non-empty suffix. When the corresponding marker exists, the input file does not enter the candidate queue again.

`cleanup.policy.maintain.relative.path`

Whether to preserve relative subdirectories from the input directory after recursive processing.

Type: boolean
Default: false
Importance: Low
Valid Values / Notes: Relevant only to recursive scanning. true preserves source-side subdirectories. false allows empty subdirectories left after files are moved or deleted to be cleaned up. The destination directory hierarchy is still created from the relative path calculated by the implementation.

`file.buffer.size.bytes`

The buffer size in bytes used when reading files.

Type: int
Default: 131072
Importance: Low
Valid Values / Notes: Must be at least 1. The Binary implementation still places the entire decompressed file into one record; this configuration is not a whole-file memory limit.

Batching, Task Assignment, and Timestamps

`batch.size`

The maximum number of records returned by one processing call.

Type: int
Default: 1000
Importance: Low
Valid Values / Notes: CSV, strongly typed JSON, line-delimited text, Avro, schema-less JSON, and ELF use it as a reference for the number of records per batch. Binary produces one whole-file record per file and does not split content according to this value. Boundary behavior differs among formats.

`task.partitioner`

The partitioning method used to assign files to tasks.

Type: string
Default: ByName
Importance: Medium
Valid Values / Notes: This version accepts only ByName. With multiple tasks, files are assigned according to a hash of the file basename; dynamic balancing by file size is not provided. Do not set the internal task.index or task.count configurations.

`timestamp.mode`

The timestamp source for CSV and strongly typed JSON records.

Type: string
Default: PROCESS_TIME
Importance: Medium
Valid Values / Notes: Valid options are FIELD, FILE_TIME, and PROCESS_TIME. Only CSV and SpoolDirJsonSourceConnector actually apply this configuration. Avro, Binary, line-delimited text, schema-less JSON, and ELF register the key but output records with null timestamps.

CSV and Strongly Typed JSON Schemas

`key.schema`

The Kafka Connect schema JSON for CSV or strongly typed JSON message keys.

Type: string
Default: Empty string
Importance: High
Valid Values / Notes: Applies only to CSV and SpoolDirJsonSourceConnector. When schema generation is disabled, it must be provided together with value.schema. A non-empty value must deserialize to a Kafka Connect schema.

`value.schema`

The Kafka Connect schema JSON for CSV or strongly typed JSON message values.

Type: string
Default: Empty string
Importance: High
Valid Values / Notes: Applies only to CSV and SpoolDirJsonSourceConnector. When schema generation is disabled, it must be provided together with key.schema. CSV columns and JSON fields are converted to strongly typed values according to this schema.

`schema.generation.enabled`

Whether to generate key and value schemas for CSV or strongly typed JSON from input samples.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Applies only to CSV and SpoolDirJsonSourceConnector. When enabled, matching files must exist at the top level of the input directory at startup. Up to five files are sampled, and the samples must generate consistent schemas. Generated fields are optional strings; numeric, Boolean, and time types are not inferred.

`schema.generation.key.fields`

The field names included in the message key during schema generation.

Type: list
Default: Empty list
Importance: Medium
Valid Values / Notes: Used only when schema generation is enabled for CSV or strongly typed JSON. The specified fields must exist in the generated value schema.

`schema.generation.key.name`

The name of the automatically generated key schema.

Type: string
Default: com.github.jcustenborder.kafka.connect.model.Key
Importance: Medium
Valid Values / Notes: Applies only to CSV and strongly typed JSON. When schema generation is enabled, it must resolve to a non-empty value.

`schema.generation.value.name`

The name of the automatically generated value schema.

Type: string
Default: com.github.jcustenborder.kafka.connect.model.Value
Importance: Medium
Valid Values / Notes: Applies only to CSV and strongly typed JSON. When schema generation is enabled, it must resolve to a non-empty value.

`parser.timestamp.timezone`

The time zone used to parse date-time text in CSV or strongly typed JSON.

Type: string
Default: UTC
Importance: Low
Valid Values / Notes: Java may silently fall back to GMT for an unknown time zone ID. Explicitly use a valid Java time zone ID.

`parser.timestamp.date.formats`

The date-time formats attempted in order when parsing timestamps in CSV or strongly typed JSON.

Type: list
Default: yyyy-MM-dd'T'HH:mm:ss,yyyy-MM-dd' 'HH:mm:ss
Importance: Low
Valid Values / Notes: Each item must be a valid SimpleDateFormat pattern. Patterns are attempted in list order; place more precise formats first.

`timestamp.field`

The field name used for field-based timestamps.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Applies only to CSV and strongly typed JSON. When timestamp.mode=FIELD, it must refer to a non-optional field in the value schema whose logical name is org.apache.kafka.connect.data.Timestamp.

CSV Parsing and Field Mapping

`csv.skip.lines`

The number of initial lines to skip before parsing CSV.

Type: int
Default: 0
Importance: Low
Valid Values / Notes: Applies only to CSV. This version does not validate a non-negative range; use a value greater than or equal to 0.

`csv.separator.char`

The Unicode numeric value of the CSV separator.

Type: int
Default: 44
Importance: Low
Valid Values / Notes: Applies only to CSV. 44 represents a comma, and 9 represents a tab. Setting it to 0 selects the RFC 4180 parser branch.

`csv.quote.char`

The Unicode numeric value of the CSV quote character.

Type: int
Default: 34
Importance: Low
Valid Values / Notes: Applies only to CSV. 34 represents a double quote.

`csv.escape.char`

The Unicode numeric value of the CSV escape character.

Type: int
Default: 92
Importance: Low
Valid Values / Notes: Applies only to the default CSV parser. 92 represents a backslash. The RFC 4180 parser ignores this configuration.

`csv.strict.quotes`

Whether the default CSV parser accepts only characters inside quotes.

Type: boolean
Default: false
Importance: Low
Valid Values / Notes: Applies only to the default CSV parser. The RFC 4180 parser ignores this configuration.

`csv.ignore.leading.whitespace`

Whether the default CSV parser ignores leading whitespace before a quote.

Type: boolean
Default: true
Importance: Low
Valid Values / Notes: Applies only to the default CSV parser. The RFC 4180 parser ignores this configuration.

`csv.ignore.quotations`

Whether the default CSV parser ignores the special meaning of quotes.

Type: boolean
Default: false
Importance: Low
Valid Values / Notes: Applies only to the default CSV parser. The RFC 4180 parser ignores this configuration.

`csv.keep.carriage.return`

Whether the CSV reader preserves carriage-return characters in lines.

Type: boolean
Default: false
Importance: Low
Valid Values / Notes: Applies only to CSV and affects both the default and RFC 4180 parser branches.

`csv.verify.reader`

Whether the CSV reader verifies the state of the underlying reader.

Type: boolean
Default: true
Importance: Low
Valid Values / Notes: Applies only to CSV and affects both the default and RFC 4180 parser branches.

`csv.null.field.indicator`

How empty delimited fields or empty quoted fields are recognized as null.

Type: string
Default: NEITHER
Importance: Low
Valid Values / Notes: Applies only to CSV. Valid options are EMPTY_SEPARATORS, EMPTY_QUOTES, BOTH, and NEITHER.

`csv.first.row.as.header`

Whether to use the first parsed record as the field-name header.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Applies only to CSV. When set to true, header names must exactly match the fields in the value schema. When set to false, columns are mapped in the field order of value.schema.

`csv.file.charset`

The character set of CSV files.

Type: string
Default: Charset.defaultCharset().name()
Importance: Low
Valid Values / Notes: Applies only to CSV, and the value must be a character set name supported by Java. Runtime reading uses this value, but schema generation still uses the JVM default character set. For non-default encodings, prefer explicit schemas and explicitly configure the default character set on the Worker.

`csv.case.sensitive.field.names`

Whether CSV field names are declared case-sensitive.

Type: boolean
Default: false
Importance: Low
Valid Values / Notes: Applies only to CSV. Although this configuration is registered and read in 2.0.71, it does not change header lookup or schema generation; actual header matching remains case-sensitive.

`csv.rfc.4180.parser.enabled`

Whether to use the RFC 4180 CSV parser.

Type: boolean
Default: false
Importance: Low
Valid Values / Notes: Applies only to CSV. When enabled, csv.escape.char, csv.strict.quotes, csv.ignore.leading.whitespace, and csv.ignore.quotations are not used to construct this parser.

Line-Delimited Text and Schema-Less JSON Character Sets

`file.charset`

The declared character set of line-delimited text or schema-less JSON files.

Type: string
Default: Charset.defaultCharset().name()
Importance: Low
Valid Values / Notes: Registered only by SpoolDirLineDelimitedSourceConnector and SpoolDirSchemaLessJsonSourceConnector. The line-delimited text implementation reads using this character set. In 2.0.71, the schema-less JSON implementation does not pass this value to the JSON parser. CSV uses csv.file.charset; the other four implementations do not register a character set configuration.

Best Practices

Avoid Reading Files That Are Still Being Written

Applicable scenario: An upstream application writes a large CSV file before handing it off to the Connector. To reduce the risk of the Connector starting to read a file before it is stable, require the file to remain unchanged for a minimum period, and have the upstream system write it using a temporary file name that does not match the input pattern before atomically renaming it to .csv when complete. Add the following to the Quick Start configuration:

file.minimum.age.ms=60000

Key considerations: The example requires the file’s last-modified time to remain unchanged for at least 60 seconds, but this configuration is checked only during discovery and does not replace atomic delivery. The upstream system can first write a file such as orders.csv.tmp, then rename it to orders.csv after writing and closing it. The temporary name must not match the input.file.pattern from the Quick Start. Adjust the minimum age according to file generation time and acceptable ingestion latency.

Isolate Bad Files and Continue Processing Later Files

Applicable scenario: A directory continuously receives multiple independent files, and a single file with invalid formatting or a schema mismatch should not stop the entire task for an extended period. Move the failed file to an isolation directory and allow the task to continue selecting other files. Add the following to the Quick Start configuration:

halt.on.error=false

Key considerations: error.path must still exist and be writable. Error handling is performed at whole-file granularity; records already sent from a file are not rolled back if a later record fails. Monitor the error directory and task logs. After correcting a file, submit it again with a new unique file name to avoid reusing the historical basename and offset.

Monitoring

What to Monitor

Monitor Kafka Connect health, Connector and Task status, throughput, latency, offset commits, errors, retries, and Worker JVM signals; monitor DLQ activity only when the corresponding error handling is enabled.

Import the Grafana Dashboard

Ensure that Connect metrics are available in a Grafana data source and that the collected labels match the dashboard filters; download the Kafka Connect dashboard, import the JSON into Grafana, and select the corresponding data source.

Limitations

The seven implementations do not share a complete set of configurations or identical record semantics. Schema, CSV, character set, and timestamp configurations must be used according to the selected connector.class; CSV behavior cannot be generalized to the other formats.
The source partition uses only the file basename and does not include the directory, topic, Connector class, or a content digest. Files with the same name in recursive directories, as well as later submissions that reuse a file name, share historical offsets and may cause records to be skipped or resent.
CSV, strongly typed JSON, and ELF resume reading from saved offsets. Line-delimited text, schema-less JSON, and Binary ignore recovered offsets and resend from the beginning of a file. Avro may duplicate the last committed record at the recovery boundary. Therefore, reliable checkpoint recovery cannot be promised uniformly across all seven implementations.
There is no atomic protocol among file cleanup, Kafka writes, and offset commits. The Connector cannot claim exactly-once processing, at-least-once processing, no duplicates, or no loss on this basis. MOVE, MOVEBYDATE, DELETE, and NONE produce different outcomes within failure windows.
When a Worker is forcibly terminated, it may leave behind a processing.file.extension marker. The plugin has no automatic expiration or startup cleanup mechanism, so a stale marker continues to prevent the corresponding file from being discovered.
The relative path calculated during recursive scanning includes the file name, so a moved file may end up at <subdirectory>/<file-name>/<file-name>. The file.relative.path header may also contain the file name rather than only the parent directory.
The Binary implementation reads the entire decompressed file into memory and produces one Kafka message. It is constrained by Worker memory, Producer request size, and Kafka message size.
timestamp.mode takes effect only for CSV and strongly typed JSON. Records created by the other five implementations have null timestamps.
When an ELF batch reaches batch.size, the implementation reads ahead to the next log entry without adding it to the result, which creates a risk of record loss at batch boundaries.
Schema generation examines only matching files at the top level of input.path and does not use recursive scanning. Samples generate only optional string fields and cannot represent the complete type constraints of later files.

FAQ

The Connector Reports That the Input, Finished, or Error Directory Is Unavailable at Startup

A task can fail to start if a directory does not exist, is not a directory, or the Connect process lacks read and write permissions. Confirm that input.path and error.path have been created and are writable. When using MOVE or MOVEBYDATE, also confirm that finished.path has been created and is writable. If a directory comes from a shared volume, inspect the mounted path and permissions from the node or container where the Worker actually runs.

The Connector Cannot Start After Schema Generation Is Enabled

Startup can fail if the top level of the input directory has no matching file, a sample file cannot be parsed, multiple samples generate inconsistent schemas, or schema.generation.key.fields names a field that does not exist. Before startup, place at least one regular file matching input.file.pattern at the top level of input.path, ensure that up to the first five samples have consistent structures, and verify the key field names. The generator does not discover samples in nested directories. For recursive reading, first start with a top-level sample, or disable schema generation and explicitly configure key.schema and value.schema.

A File in the Input Directory Is Never Processed

Common causes include a file name that does not fully match input.file.pattern, a file younger than file.minimum.age.ms, a file in a subdirectory when recursive scanning is disabled, or a stale processing marker for the same file name. Check the actual file name, its last-modified time, and input.path.walk.recursively, and look for <file-name><processing.file.extension>. Remove a stale marker manually only after confirming that no task is processing the file.

Why Are Records Duplicated or Skipped After a Restart?

Offset recovery differs among formats, and the source partition is identified only by the basename. Line-delimited text, schema-less JSON, and Binary reread rediscovered files from the beginning. A new file with a reused name, or files with the same name in different subdirectories, also reuse old offsets. Use a unique basename for every submission, retain the original files in the finished directory for reconciliation, and do not replace a same-named file with new content within the same Connector offset namespace. For pipelines where duplicates or missing records are unacceptable, use stable business keys for idempotent downstream processing and integrity checks.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Output and File Discovery

​topic

​input.path

​input.file.pattern

​file.minimum.age.ms

​input.path.walk.recursively

​files.sort.attributes

​empty.poll.wait.ms

​File Lifecycle and Error Handling

​cleanup.policy

​finished.path

​error.path

​halt.on.error

​processing.file.extension

​cleanup.policy.maintain.relative.path

​file.buffer.size.bytes

​Batching, Task Assignment, and Timestamps

​batch.size

​task.partitioner

​timestamp.mode

​CSV and Strongly Typed JSON Schemas

​key.schema

​value.schema

​schema.generation.enabled

​schema.generation.key.fields

​schema.generation.key.name

​schema.generation.value.name

​parser.timestamp.timezone

​parser.timestamp.date.formats

​timestamp.field

​CSV Parsing and Field Mapping

​csv.skip.lines

​csv.separator.char

​csv.quote.char

​csv.escape.char

​csv.strict.quotes

​csv.ignore.leading.whitespace

​csv.ignore.quotations

​csv.keep.carriage.return

​csv.verify.reader

​csv.null.field.indicator

​csv.first.row.as.header

​csv.file.charset

​csv.case.sensitive.field.names

​csv.rfc.4180.parser.enabled

​Line-Delimited Text and Schema-Less JSON Character Sets

​file.charset

​Best Practices

​Avoid Reading Files That Are Still Being Written

​Isolate Bad Files and Continue Processing Later Files

​Monitoring

​What to Monitor

​Import the Grafana Dashboard

​Limitations

​FAQ

​The Connector Reports That the Input, Finished, or Error Directory Is Unavailable at Startup

​The Connector Cannot Start After Schema Generation Is Enabled

​A File in the Input Directory Is Never Processed

​Why Are Records Duplicated or Skipped After a Restart?