<a id="spooldir-binary-source-connector"></a>

# Binary File Source Connector for Confluent Platform

The Binary File Source connector is used to read an entire file as a byte array
and write the data to Kafka. To use this connector, use a connector configuration
that specifies the name of the connector class in the `connector.class`
configuration property:

```text
connector.class=com.github.jcustenborder.kafka.connect.spooldir.SpoolDirBinaryFileSourceConnector
```

#### IMPORTANT
- Large files will be read as a single byte array. This means that the
  process could run out of memory or try to send a message to Kafka that is
  greater than the maximum message size. If this happens, the connector will
  throw an exception.
- The recommended converter to use is the ByteArrayConverter (for example,
  `value.converter=org.apache.kafka.connect.storage.ByteArrayConverter`)

## Configuration Properties

The Binary File Source connector can be configured using a variety of configuration properties.

### General

The following are general configuration properties.

#### topic

The Kafka topic to write the data to.

- Importance: HIGH
- Type: STRING

#### batch.size

The number of records that should be returned with each batch.

- Importance: LOW
- Type: INT
- Default value: 1000

#### empty.poll.wait.ms

The amount of time, in milliseconds (ms), to wait if a poll returns an empty list of records.

- Importance: LOW
- Type: LONG
- Default value: 500
- Valid values: [1,…,9223372036854775807]

#### task.count

An internal setting to the connector used to instruct a task on which files to
select. The connector will override this setting.

- Importance: LOW
- Type: INT
- Default value: 1
- Valid values: [1,…]

#### task.index

An internal setting used by the Spool Dir Connector to select which files are
processed by each task. The connector will override this setting.

- Importance: LOW
- Type: INT
- Default value: 0
- Valid values: [0,…]

### File System

Use the following properties to configure your file system.

#### error.path

The directory to place files of which has error(s). This directory must exist
and be writable by the user running Kafka Connect.

- Importance: HIGH
- Type: STRING
- Valid values: Absolute path to a directory that exists and is writable.

#### input.file.pattern

Regular expression to check input filenames against. This expression must
match the entire filename. The equivalent of `Matcher.matches()`.

- Importance: HIGH
- Type: STRING

#### input.path

The directory to read files from which will be processed. This directory must
exist and be writable by the user running Kafka Connect.

- Importance: HIGH
- Type: STRING
- Valid values: Absolute path to a directory that exists and is writable.

#### finished.path

The directory to place files that have been processed. This directory must exist
and be writable by the user running Kafka Connect .

- Importance: HIGH
- Type: STRING

#### halt.on.error

Determines whether the task should halt when it encounters an error or continue
to the next file.

- Importance: HIGH
- Type: BOOLEAN
- Default value: true

#### cleanup.policy

Determines how the connector should cleanup the files that have been processed.
A value of `NONE` leaves the files in place which causes them to be
reprocessed if the connector is restarted. `DELETE` removes the file from the
filesystem. `MOVE` will move the file to a finished directory. `MOVEBYDATE` will
move the file to a finished directory with subdirectories by date

- Importance: MEDIUM
- Type: STRING
- Default value: MOVE
- Valid values: `NONE`, `DELETE`, `MOVE`, `MOVEBYDATE`

#### task.partitioner

The task partitioner implementation is used when the connector is configured to
use more than one task. This is used by each task to identify which files will
be processed by that task. This ensures that each file is only assigned to one
task.

- Importance: MEDIUM
- Type: STRING
- Default value: ByName
- Valid values: `ByName`

#### file.buffer.size.bytes

The size of buffer for the BufferedInputStream that will be used to interact
with the file system.

- Importance: LOW
- Type: INT
- Default value: 131072
- Valid values: [1,…]

#### file.minimum.age.ms

The amount of time in milliseconds after the file was last written to before the
file can be processed.

- Importance: LOW
- Type: LONG
- Default value: 0
- Valid values: [0,…]

#### files.sort.attributes

The attributes each file will use to determine the sort order. `Name` is name of
the file. `Length` is the length of the file preferring larger files first.
`LastModified` is the LastModified attribute of the file preferring older files
first.

- Importance: LOW
- Type: LIST
- Default value: [NameAsc]
- Valid values: `NameAsc`, `NameDesc`, `LengthAsc`, `LengthDesc`, `LastModifiedAsc`, `LastModifiedDesc`

#### processing.file.extension

Before a file is processed, a flag is created in its directory to indicate the
file is being handled. The flag file has the same name as the file, but with
this property appended as a suffix.

- Importance: LOW
- Type: STRING
- Default value: `.PROCESSING`
- Valid values: regex( ^.\*..+$ )

### Timestamps

Use the following properties for configuring the timestamp.

#### timestamp.mode

Determines how the connector will set the timestamp for the
[ConnectRecord]([https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/connector/ConnectRecord](https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/connector/ConnectRecord).html#timestamp()).
If set to `Field` then the timestamp will be read from a field in the value.
This field cannot be optional and must be a
[Timestamp]([https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.html](https://kafka.apache.org/0102/javadoc/org/apache/kafka/connect/data/Schema.html)).
Specify the field  in `timestamp.field`. If set to `FILE_TIME` then the last
modified time of the file will be used. If set to `PROCESS_TIME` the time the
record is read will be used.

- Importance: MEDIUM
- Type: STRING
- Default value: PROCESS_TIME
- Valid values:: `FIELD`, `FILE_TIME`, `PROCESS_TIME`

<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

### CSFLE and CSPE configurations

`csfle.enabled`

Accepts a boolean value. CSFLE is enabled for the connector if `csfle.enabled` is set to True.

* Type: boolean
* Default: False

`auto.register.schemas`

Specifies if the Serializer should attempt to register the Schema with Schema Registry.

* Type: boolean
* Default: true
* Importance: medium

`use.latest.version`

Only applies when `auto.register.schemas` is set to false. If `auto.register.schemas` is set to false and `use.latest.version` is set to true, then instead of deriving a schema for the object passed to the client for serialization, Schema Registry uses the latest version of the schema in the subject for serialization.

* Type: boolean
* Default: true
* Importance: medium