> ## 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.

# Neo4j Sink Connector

> Configure and operate the Neo4j Sink Connector in AutoMQ Connect, including Cypher and pattern strategies, delivery behavior, security, monitoring, and troubleshooting.

## Overview

The Neo4j Sink Connector consumes records from Kafka topics and converts them into writes of nodes, relationships, and properties to Neo4j or Aura databases. The connector sits between Kafka and the graph database and supports selecting a Cypher, Pattern, CUD, CDC Schema, or CDC Source Id strategy for each topic. Each topic must be associated with exactly one strategy.

The Cypher strategy is suitable when custom statements are needed to control graph writes. The Pattern strategy is suitable for declaratively mapping fields to nodes or relationships. The CUD and CDC strategies consume their corresponding operation or change event formats. Typical uses include building knowledge graphs from business events, continuously updating entity relationships, and transferring CDC events between Neo4j databases.

## License

Licensed under the Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka, and a target Neo4j database, and confirm network connectivity and access permissions. For preparation and management procedures, see [Manage Connectors](../manage-connectors).

```properties theme={null}
connector.class=org.neo4j.connectors.kafka.sink.Neo4jConnector
topics=people
neo4j.uri=neo4j://<neo4j-host>:7687
neo4j.database=<neo4j-database>
neo4j.authentication.type=BASIC
neo4j.authentication.basic.username=<neo4j-username>
neo4j.authentication.basic.password=<neo4j-password>
neo4j.cypher.topic.people=MERGE (p:Person {id: __value.id}) SET p.name = __value.name, p.surname = __value.surname
```

Replace the Neo4j address, database, and credential placeholders. The example requires message values in the `people` topic to provide the `id`, `name`, and `surname` fields. The connector executes Cypher using the default `__value` binding and merges by `id` into `Person` nodes. In production, provide the password through a secure credential management mechanism.

## Configuration

### Connector Identity and Topic Subscription

#### `connector.class`

The Neo4j Sink Connector implementation class to load.

* **Type**: `string`
* **Default**: None
* **Importance**: High
* **Valid values / Notes**: Use `org.neo4j.connectors.kafka.sink.Neo4jConnector`.
* **Required**: Yes

#### `topics`

The list of Kafka topics to consume.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: High
* **Valid values / Notes**: Use a comma-separated list of literal topic names. Each name must be handled by exactly one sink strategy. Neo4j 5.5.1 cannot use `topics.regex` to establish the required per-topic strategy mapping.
* **Required**: Yes

### Neo4j Connection

#### `neo4j.uri`

One or more Neo4j connection URIs.

* **Type**: `list`
* **Default**: None
* **Importance**: High
* **Valid values / Notes**: The URI scheme must be `neo4j`, `neo4j+s`, `neo4j+ssc`, `bolt`, `bolt+s`, or `bolt+ssc`. Multiple URIs are used for custom address resolution. If the port is omitted, `7687` is used.
* **Required**: Yes

#### `neo4j.database`

The name of the Neo4j database to write to.

* **Type**: `string`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: An empty string lets the Neo4j driver or server select the database. Explicitly specifying the target database is recommended.

### Authentication

#### `neo4j.authentication.type`

The authentication type used to connect to Neo4j.

* **Type**: `string`
* **Default**: `BASIC`
* **Importance**: High
* **Valid values / Notes**: Valid values are `NONE`, `BASIC`, `KERBEROS`, `BEARER`, and `CUSTOM`. The selected type determines which credential settings must be provided.

#### `neo4j.authentication.basic.username`

The Basic authentication username.

* **Type**: `string`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Must be non-empty when `neo4j.authentication.type=BASIC`.
* **Required**: `conditional` (required when using Basic authentication)

#### `neo4j.authentication.basic.password`

The Basic authentication password.

* **Type**: `password`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Must be non-empty when `neo4j.authentication.type=BASIC`. Provide it through a Config Provider or an equivalent secret management mechanism.
* **Required**: `conditional` (required when using Basic authentication)

#### `neo4j.authentication.basic.realm`

The Basic authentication realm.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Low
* **Valid values / Notes**: Used only for Basic authentication. An empty string selects the default realm.

#### `neo4j.authentication.kerberos.ticket`

The Kerberos authentication ticket.

* **Type**: `password`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Must be non-empty when `neo4j.authentication.type=KERBEROS` and should be provided through a secure credential management mechanism.
* **Required**: `conditional` (required when using Kerberos authentication)

#### `neo4j.authentication.bearer.token`

The Bearer authentication token.

* **Type**: `password`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Must be non-empty when `neo4j.authentication.type=BEARER` and should be provided through a secure credential management mechanism.
* **Required**: `conditional` (required when using Bearer authentication)

#### `neo4j.authentication.custom.scheme`

The custom authentication scheme.

* **Type**: `string`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Must be non-empty when `neo4j.authentication.type=CUSTOM`.
* **Required**: `conditional` (required when using custom authentication)

#### `neo4j.authentication.custom.principal`

The custom authentication principal.

* **Type**: `string`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Must be non-empty when `neo4j.authentication.type=CUSTOM`.
* **Required**: `conditional` (required when using custom authentication)

#### `neo4j.authentication.custom.credentials`

The custom authentication credentials.

* **Type**: `password`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Must be non-empty when `neo4j.authentication.type=CUSTOM` and should be provided through a secure credential management mechanism.
* **Required**: `conditional` (required when using custom authentication)

#### `neo4j.authentication.custom.realm`

The realm passed to the custom Neo4j Auth Token.

* **Type**: `string`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: Used only for custom authentication. Whether it is required depends on the authentication provider.

### TLS

#### `neo4j.security.encrypted`

Whether to enable plugin-managed encryption for plain `bolt` or `neo4j` URIs.

* **Type**: `string` (logical type: boolean)
* **Default**: `false`
* **Importance**: Low
* **Valid values / Notes**: Use only `true` or `false`. Encryption for `+s` and `+ssc` URIs is controlled by the URI itself and does not use this setting.

#### `neo4j.security.trust-strategy`

The certificate trust strategy used with plugin-managed TLS.

* **Type**: `string`
* **Default**: `TRUST_SYSTEM_CA_SIGNED_CERTIFICATES`
* **Importance**: Low
* **Valid values / Notes**: Valid values are `TRUST_ALL_CERTIFICATES`, `TRUST_CUSTOM_CA_SIGNED_CERTIFICATES`, and `TRUST_SYSTEM_CA_SIGNED_CERTIFICATES`. This setting applies only when plugin-managed encryption is enabled for a plain URI.
* **Required**: `conditional` (must be non-empty when plugin-managed encryption is enabled)

#### `neo4j.security.hostname-verification-enabled`

Whether to verify the certificate hostname during the TLS handshake.

* **Type**: `string` (logical type: boolean)
* **Default**: `true`
* **Importance**: Low
* **Valid values / Notes**: Use only `true` or `false`. This setting applies only to plugin-managed encryption enabled for a plain URI.
* **Required**: `conditional` (must be non-empty when plugin-managed encryption is enabled)

#### `neo4j.security.cert-files`

The list of custom CA certificate files.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: Low
* **Valid values / Notes**: Each entry must be an absolute path to a regular file readable by the worker. The list must not be empty when `TRUST_CUSTOM_CA_SIGNED_CERTIFICATES` is selected.
* **Required**: `conditional` (required when using the custom CA trust strategy)

### Connection Pool and Transaction Retries

#### `neo4j.connection-timeout`

The timeout for establishing a Neo4j TCP connection.

* **Type**: `string` (logical type: duration)
* **Default**: `30s`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative compound duration using `ms`, `s`, `m`, `h`, or `d`.

#### `neo4j.pool.max-connection-pool-size`

The maximum number of connections retained by the Neo4j driver connection pool.

* **Type**: `int`
* **Default**: `100`
* **Importance**: Low
* **Valid values / Notes**: Must be at least `1`. Adjust it based on task parallelism and target database capacity.

#### `neo4j.pool.connection-acquisition-timeout`

The maximum time to wait for a connection from the connection pool.

* **Type**: `string` (logical type: duration)
* **Default**: `1m`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative compound duration using `ms`, `s`, `m`, `h`, or `d`.

#### `neo4j.pool.max-connection-lifetime`

The maximum lifetime of a connection in the pool.

* **Type**: `string` (logical type: duration)
* **Default**: `1h`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative compound duration using `ms`, `s`, `m`, `h`, or `d`.

#### `neo4j.pool.idle-time-before-connection-test`

The amount of time an idle connection may remain idle before a liveness check is performed.

* **Type**: `string` (logical type: duration)
* **Default**: `""`
* **Importance**: Low
* **Valid values / Notes**: May be empty or a non-negative compound duration. Leaving it empty preserves the driver's disabled or default liveness-check semantics.

#### `neo4j.max-retry-time`

The maximum time the Neo4j driver retries retryable transactions.

* **Type**: `string` (logical type: duration)
* **Default**: `30s`
* **Importance**: Low
* **Valid values / Notes**: Must be a non-negative compound duration using `ms`, `s`, `m`, `h`, or `d`. It does not control Kafka Connect framework error tolerance or DLQ behavior.

### Write Strategies and Topic Routing

#### `neo4j.cypher.topic.<topic>`

The Cypher statement executed for the specified topic.

* **Type**: `string`
* **Default**: None
* **Importance**: N/A
* **Valid values / Notes**: Replace `<topic>` with the exact topic name from `topics`. Each topic can use only one strategy. Cypher is parsed and executed at task runtime.
* **Required**: `conditional` (required when the topic uses the Cypher strategy)

#### `neo4j.pattern.topic.<topic>`

The node or relationship Pattern mapping for the specified topic.

* **Type**: `string`
* **Default**: None
* **Importance**: N/A
* **Valid values / Notes**: Replace `<topic>` with the exact topic name from `topics`. The Pattern must parse as a node or relationship mapping, and the topic cannot use another strategy.
* **Required**: `conditional` (required when the topic uses the Pattern strategy)

#### `neo4j.cud.topics`

The list of topics that use the CUD operation format.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: Medium
* **Valid values / Notes**: Topics in the list must also appear in `topics` and cannot be assigned to another strategy. Message values must conform to the connector's CUD structure.

#### `neo4j.cdc.schema.topics`

The list of topics that use the CDC Schema strategy.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: Medium
* **Valid values / Notes**: Topics in the list must also appear in `topics` and cannot be assigned to another strategy. Messages must use a supported Neo4j CDC Schema or a compatible legacy Streams event structure.

#### `neo4j.cdc.source-id.topics`

The list of topics that use the CDC Source Id strategy.

* **Type**: `list`
* **Default**: Empty list
* **Importance**: Medium
* **Valid values / Notes**: Topics in the list must also appear in `topics` and cannot be assigned to another strategy. Events must contain a source element ID.

#### `neo4j.cdc.source-id.label-name`

The identity label used for target nodes by the CDC Source Id strategy.

* **Type**: `string`
* **Default**: `SourceEvent`
* **Importance**: Low
* **Valid values / Notes**: Used only when `neo4j.cdc.source-id.topics` is non-empty. Keep it stable for the source to avoid collisions between element IDs from different sources.

#### `neo4j.cdc.source-id.property-name`

The property name used by the CDC Source Id strategy to store the source element ID.

* **Type**: `string`
* **Default**: `sourceId`
* **Importance**: Low
* **Valid values / Notes**: Used only when `neo4j.cdc.source-id.topics` is non-empty. Changing the property name changes how target identities are matched.

### Cypher Record Bindings

#### `neo4j.cypher.bind-timestamp-as`

The variable name that exposes the Kafka record timestamp in Cypher.

* **Type**: `string`
* **Default**: `__timestamp`
* **Importance**: Medium
* **Valid values / Notes**: An empty string disables the binding. The timestamp is converted to a UTC Offset Date-Time. At least one valid Cypher binding must remain when a binding is disabled.

#### `neo4j.cypher.bind-header-as`

The variable name that exposes Kafka headers in Cypher.

* **Type**: `string`
* **Default**: `__header`
* **Importance**: Low
* **Valid values / Notes**: An empty string disables the binding. At least one valid Cypher binding must remain when a binding is disabled.

#### `neo4j.cypher.bind-key-as`

The variable name that exposes the Kafka record key in Cypher.

* **Type**: `string`
* **Default**: `__key`
* **Importance**: Low
* **Valid values / Notes**: An empty string disables the binding. At least one valid Cypher binding must remain when a binding is disabled.

#### `neo4j.cypher.bind-value-as`

The variable name that exposes the Kafka record value in Cypher.

* **Type**: `string`
* **Default**: `__value`
* **Importance**: Low
* **Valid values / Notes**: An empty string disables the binding. At least one valid Cypher binding must remain when a binding is disabled.

#### `neo4j.cypher.bind-value-as-event`

Whether to also bind the record value using the legacy `event` variable name.

* **Type**: `string` (logical type: boolean)
* **Default**: `true`
* **Importance**: Low
* **Valid values / Notes**: Use only `true` or `false`. When set to `false`, at least one explicit Cypher binding must remain non-empty.

### Pattern Record Bindings and Property Writes

#### `neo4j.pattern.bind-timestamp-as`

The Kafka record timestamp alias used in Pattern expressions.

* **Type**: `string`
* **Default**: `__timestamp`
* **Importance**: Low
* **Valid values / Notes**: An empty string disables the alias.

#### `neo4j.pattern.bind-header-as`

The Kafka header alias used in Pattern expressions.

* **Type**: `string`
* **Default**: `__header`
* **Importance**: Low
* **Valid values / Notes**: An empty string disables the alias.

#### `neo4j.pattern.bind-key-as`

The Kafka record key alias used in Pattern expressions.

* **Type**: `string`
* **Default**: `__key`
* **Importance**: Low
* **Valid values / Notes**: An empty string disables the alias.

#### `neo4j.pattern.bind-value-as`

The Kafka record value alias used in Pattern expressions.

* **Type**: `string`
* **Default**: `__value`
* **Importance**: Low
* **Valid values / Notes**: An empty string disables the alias. The value of a non-tombstone message must be convertible to a Map.

#### `neo4j.pattern.merge-node-properties`

Whether the Pattern strategy includes incoming properties in node `MERGE` matching conditions.

* **Type**: `string` (logical type: boolean)
* **Default**: `false`
* **Importance**: Low
* **Valid values / Notes**: Use only `true` or `false`. This setting applies only to node and relationship Pattern strategies. When enabled, property changes may affect node matching results.

#### `neo4j.pattern.merge-relationship-properties`

Whether the relationship Pattern strategy includes incoming properties in relationship `MERGE` matching conditions.

* **Type**: `string` (logical type: boolean)
* **Default**: `false`
* **Importance**: Low
* **Valid values / Notes**: Use only `true` or `false`. This setting applies only to the relationship Pattern strategy. When enabled, property changes may affect relationship matching results.

### Batching and Target-side Offsets

#### `neo4j.batch-size`

The maximum number of events included when generating a batch of write statements for each topic.

* **Type**: `int`
* **Default**: `1000`
* **Importance**: Medium
* **Valid values / Notes**: Must be at least `1`. When APOC is available, it is used to split write transaction batches. On the native path, it limits the number of events in each generated statement, but multiple statements for the same topic group may still be in one Neo4j managed transaction.

#### `neo4j.batch-timeout`

The publicly registered batch duration setting.

* **Type**: `string` (logical type: duration)
* **Default**: `0s`
* **Importance**: Medium
* **Valid values / Notes**: Accepts non-negative compound durations using `ms`, `s`, `m`, `h`, or `d`. The main sink execution path in Neo4j Connector 5.5.1 does not read this setting, so it cannot be relied on to limit batch or transaction execution time.

#### `neo4j.max-batched-queries`

The maximum number of distinct statement shapes that can be combined in a batch on the native batching path.

* **Type**: `int`
* **Default**: `50`
* **Importance**: Low
* **Valid values / Notes**: Must be at least `1`. When the limit is reached, the native path splits the generated work. The APOC path does not read this setting.

#### `neo4j.eos-offset-label`

The Neo4j label used for target-side offset nodes.

* **Type**: `string`
* **Default**: `""`
* **Importance**: High
* **Valid values / Notes**: An empty string disables target-side offset nodes. When non-empty, the connector stores written offsets by strategy, topic, and partition, and requires the corresponding constraint in the target database. This mechanism only filters offsets committed to Neo4j within the same identity scope and does not provide an end-to-end or global exactly-once guarantee. Do not rely on this mechanism when the same task processes records from multiple partitions of the same topic in one operation.

### Tasks and Converters

#### `tasks.max`

The maximum number of tasks the connector requests to create.

* **Type**: `int`
* **Default**: `1`
* **Importance**: High
* **Valid values / Notes**: Must be at least `1`. Actual parallelism is limited by Kafka partition assignment and Neo4j concurrency capacity. There is no global ordering across tasks, topics, or partitions.

#### `tasks.max.enforce`

Whether to enforce that the number of tasks returned by the connector does not exceed `tasks.max`.

* **Type**: `boolean`
* **Default**: `true`
* **Importance**: Low
* **Valid values / Notes**: Kafka Connect 3.9.1 has deprecated this setting and plans to remove it in a future major release. No replacement is provided. Keep it set to `true`.
* **Deprecated**: Yes

#### `key.converter`

The connector-level Kafka record key converter class.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: When omitted, the worker's key converter is inherited. An explicit value must be an instantiable Kafka Connect `Converter` implementation. Converter-specific settings are defined by the corresponding plugin.

#### `value.converter`

The connector-level Kafka record value converter class.

* **Type**: `class`
* **Default**: `null`
* **Importance**: Low
* **Valid values / Notes**: When omitted, the worker's value converter is inherited. An explicit value must be an instantiable Kafka Connect `Converter` implementation. All sink strategies use the converted value.

### Error Handling and DLQ

#### `errors.tolerance`

The Kafka Connect tolerance policy for record errors.

* **Type**: `string`
* **Default**: `none`
* **Importance**: Medium
* **Valid values / Notes**: Valid values are `none` and `all`. `all` allows the Errant Record Reporter to isolate records that cannot be processed, but setting this option alone does not configure a DLQ. This mechanism is separate from the database transaction retries controlled by `neo4j.max-retry-time`.

#### `errors.deadletterqueue.topic.name`

The name of the DLQ topic for records that cannot be processed.

* **Type**: `string`
* **Default**: `""`
* **Importance**: Medium
* **Valid values / Notes**: An empty string disables the DLQ. A non-empty topic must not be the same as a subscribed business topic. Use it with `errors.tolerance=all`.

#### `errors.deadletterqueue.topic.replication.factor`

The replication factor used when Kafka Connect creates the DLQ topic.

* **Type**: `short`
* **Default**: `3`
* **Importance**: Medium
* **Valid values / Notes**: Must be suitable for the number of brokers in the Kafka cluster. It is not used to create a topic when no DLQ is configured or the topic already exists.

#### `errors.deadletterqueue.context.headers.enable`

Whether to add error context headers to DLQ records.

* **Type**: `boolean`
* **Default**: `false`
* **Importance**: Medium
* **Valid values / Notes**: Applies only when a record is actually written to the DLQ. Set it to `true` to retain the connector error context needed for troubleshooting.

## Best Practices

### Declare Field-to-Graph Mappings During Initial Integration

Applicable scenario: You are writing structurally stable entity events to Neo4j for the first time, want to avoid maintaining custom Cypher, and need to define which fields identify nodes and which fields become node properties. The following configuration uses the Pattern strategy, with `id` as the matching key for `Person` nodes and `name` and `surname` written as properties.

```properties theme={null}
connector.class=org.neo4j.connectors.kafka.sink.Neo4jConnector
topics=people
neo4j.uri=neo4j://<neo4j-host>:7687
neo4j.database=<neo4j-database>
neo4j.authentication.type=BASIC
neo4j.authentication.basic.username=<neo4j-username>
neo4j.authentication.basic.password=<neo4j-password>
neo4j.pattern.topic.people=(:Person{!id,name,surname})
```

Key considerations: First ensure that the message value can be converted to a Map containing `id`, `name`, and `surname`, and create an appropriate constraint for the business unique key in the target database. In the Pattern, `!id` marks `id` as the node matching key, while `name` and `surname` are written as properties. Without valid uniqueness, the connector may match or create unintended entities. Do not configure Cypher, CUD, or CDC strategies for the same `people` topic.

### Scale Tasks by Partition for Continuous Writes

Applicable scenario: The connector is continuously writing to Neo4j, a single task cannot meet throughput or latency requirements, and the business topic has multiple partitions available for parallel assignment. The following configuration raises the task limit to `4` based on the Quick Start configuration.

```properties theme={null}
connector.class=org.neo4j.connectors.kafka.sink.Neo4jConnector
tasks.max=4
topics=people
neo4j.uri=neo4j://<neo4j-host>:7687
neo4j.database=<neo4j-database>
neo4j.authentication.type=BASIC
neo4j.authentication.basic.username=<neo4j-username>
neo4j.authentication.basic.password=<neo4j-password>
neo4j.cypher.topic.people=MERGE (p:Person {id: __value.id}) SET p.name = __value.name, p.surname = __value.surname
```

Key considerations: `tasks.max=4` is only the requested task limit and does not guarantee that four tasks will run. Effective parallelism cannot exceed the number of assignable partitions and is also limited by the Neo4j connection pool and database write capacity. Observe task assignments, throughput, latency, and database load before and after scaling. Rely only on ordering within a single topic partition; do not assume global ordering across tasks or partitions.

### Isolate Bad Records with a DLQ During Routine Operation

Applicable scenario: During continuous writes, a small number of records may fail conversion, violate a strategy's structure, or cause Neo4j writes to fail. You want to retain these records for investigation while allowing processable records to continue. The following configuration enables error tolerance, a DLQ, and context headers.

```properties theme={null}
connector.class=org.neo4j.connectors.kafka.sink.Neo4jConnector
topics=people
neo4j.uri=neo4j://<neo4j-host>:7687
neo4j.database=<neo4j-database>
neo4j.authentication.type=BASIC
neo4j.authentication.basic.username=<neo4j-username>
neo4j.authentication.basic.password=<neo4j-password>
neo4j.cypher.topic.people=MERGE (p:Person {id: __value.id}) SET p.name = __value.name, p.surname = __value.surname
errors.tolerance=all
errors.deadletterqueue.topic.name=<dlq-topic-name>
errors.deadletterqueue.topic.replication.factor=3
errors.deadletterqueue.context.headers.enable=true
```

Key considerations: The DLQ topic must not overlap with `people`, and the replication factor must be suitable for the Kafka cluster. After a batch fails, the connector attempts to isolate individual records. Successful records may already have been committed, while records accepted by the reporter and written to the DLQ are not written to Neo4j. In that case, `put` can return successfully and later Kafka offset commits may advance past those records. Continuously monitor the DLQ, correct the data or mapping, and replay records separately according to business requirements. This configuration does not provide transaction retries or an exactly-once guarantee.

## 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

Confirm that Connect metrics are available in a Grafana data source and that collected labels satisfy 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

* Sink strategies in Neo4j Connector 5.5.1 depend on exact topic names in `topics`; `topics.regex` cannot replace the literal topic list.
* Each topic must be assigned to exactly one of the Cypher, Pattern, CUD, CDC Schema, or CDC Source Id strategies. The connector does not support selecting a strategy dynamically per record or using a fallback strategy.
* Although `neo4j.batch-timeout` passes configuration validation, the main sink execution path in 5.5.1 does not read it, so it cannot limit batch or transaction execution time.
* The connector does not provide global write ordering across tasks, topics, or Kafka partitions. CDC transaction IDs and sequence numbers also do not reconstruct source transaction boundaries.
* `neo4j.eos-offset-label` filters offsets committed to Neo4j only within a stable strategy, topic, partition, target database, and label identity. It does not provide an end-to-end or global exactly-once guarantee, and should not be relied on when records from multiple partitions of the same topic enter one task batch.

## FAQ

### A Task Reports That a Topic Has No Strategy or Has Multiple Strategies at Startup

An inconsistency between `topics` and the strategy settings prevents the task from starting. Check that each literal topic appears in exactly one place: its corresponding `neo4j.cypher.topic.<topic>` or `neo4j.pattern.topic.<topic>`, or one of `neo4j.cud.topics`, `neo4j.cdc.schema.topics`, and `neo4j.cdc.source-id.topics`. Remove duplicate assignments and extra strategy topics not listed in `topics`, then restart the connector.

### The Connector Cannot Connect to or Authenticate with Neo4j

A mismatch in the URI scheme, database name, authentication type, or credentials can cause connection failures. First confirm that `neo4j.uri` is reachable from the worker, then verify the non-empty credentials required by the selected authentication type. When using TLS, also check the URI scheme, trust strategy, hostname verification, and the absolute paths and read permissions of custom CA files.

### Duplicate Nodes or Relationships Appear After a Restart

Kafka offset commits may occur after Neo4j transactions have already committed, so non-idempotent Cypher or CUD `CREATE` operations may produce duplicate results during replay. Prefer idempotent writes designed with stable business keys, constraints, and `MERGE`. If you evaluate using `neo4j.eos-offset-label`, you must also keep the strategy, topic, partition, target database, and label unchanged. The mechanism is not globally exactly-once and should not be used as a replay-filtering guarantee when one task processes records from multiple partitions of the same topic in a single operation.

### Bad Records Do Not Enter the DLQ or the Task Still Fails

Setting only the DLQ topic does not enable error tolerance. Confirm that `errors.tolerance=all` is also set, that the DLQ topic does not overlap with business subscriptions, and that Kafka Connect can access or create the topic. Then check the DLQ replication factor, reporter errors, and Neo4j exceptions. Errors that the reporter cannot accept still cause `put` to fail.