> ## Documentation Index > Fetch the complete documentation index at: https://docs.automq.com/llms.txt > Use this file to discover all available pages before exploring further. # Spooldir Source Connector > Configure and operate the Spooldir Source Connector in AutoMQ Connect, including prerequisites, configuration, monitoring, and troubleshooting. ## 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](../manage-connectors). ```properties theme={null} connector.class=com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector topic= input.path= finished.path= error.path= 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 ``. 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 `` after successful processing under the default `MOVE` policy. Files that cannot be processed are moved to ``. ## 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: ```properties theme={null} 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: ```properties theme={null} 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](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), 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 `//`. 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 ``. 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.