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

> Configure and operate the Neo4j Source Connector in AutoMQ Connect, including query polling, CDC, offsets, security, monitoring, and troubleshooting.

## Overview

Neo4j Source Connector sends Neo4j graph data to Kafka for downstream search, analytics, and event-processing applications. QUERY mode periodically executes Cypher, writes each result row to a single Topic, and tracks its read position with an integer cursor from the results. CDC mode reads the database change log and routes matching node and relationship changes to configured Topics, including create, update, and delete events.

Use QUERY to export data through a business query and then continue reading records after the cursor; use CDC to stream graph changes. The two modes interpret the earliest starting position differently: QUERY uses a numeric cursor starting point, while CDC uses the beginning of retained change history. Even when the earliest position is selected, CDC does not export all existing graph data.

## Prerequisites

* The Neo4j account used for QUERY must be able to read the data involved in the query.
* QUERY queries must return a non-null Long cursor and use a unique, monotonically increasing cursor design to support pagination and subsequent reads.
* CDC requires a Neo4j deployment that supports CDC, change capture enabled on the target database, and an account with permission to read the change log.

## License

Licensed under Apache License 2.0.

## Quick Start

Prepare a Connect Cluster, Kafka, a Neo4j database, and the target Topic, and confirm network connectivity and access permissions. For creation and management operations, see [Manage Connectors](../manage-connectors). The following configuration reads existing records from a numeric starting point and continues polling for new records.

```properties theme={null}
connector.class=org.neo4j.connectors.kafka.source.Neo4jConnector
neo4j.uri=<neo4j-uri>
neo4j.authentication.basic.username=<neo4j-username>
neo4j.authentication.basic.password=<neo4j-password>
neo4j.start-from=EARLIEST
neo4j.query=MATCH (e:Event) WHERE e.sequence > $lastCheck RETURN e.id AS id, e.sequence AS timestamp, e.payload AS payload ORDER BY timestamp
neo4j.query.topic=<query-topic>
```

Replace the connection address, account credentials, and Topic. Source `Event` nodes must contain a business identifier `id`, a Long `sequence`, and the `payload` to send. The `sequence` must be nonnegative and unique, and subsequent writes must not use sequence numbers that have already been read. This field is returned under the default cursor name `timestamp`, but represents a sequence number rather than time here. Authentication defaults to BASIC, and the driver selects the database by default; set `neo4j.database` to specify a database. The default is one Task, and Key and Value serialization uses the Worker's Converters.

When no Offset is available on the first start, `EARLIEST` sets `$lastCheck` to `-1`; an available Offset takes precedence for recovery. The query does not provide a consistent graph snapshot and cannot automatically observe hard deletes. The application must generate a new sequence number for each change that needs to be transmitted. To retain every change, write separate event records rather than repeatedly overwriting the same node.

## Configuration

### Connection and Database

#### `neo4j.uri`

Configures the Neo4j connection addresses.

* **Type**: `LIST`
* **Default**: None, required
* **Importance**: High
* **Valid Values / Notes**: A non-empty URI list, with multiple addresses separated by commas. Supports `neo4j`, `neo4j+s`, `neo4j+ssc`, `bolt`, `bolt+s`, and `bolt+ssc`. The first URI is used to create the driver. When multiple URIs are configured, all addresses are included in the driver's address resolver.

#### `neo4j.database`

Selects the database to read.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: An empty or whitespace-only value uses the driver's default database selection. Changing the database changes the source Offset identity.

### Authentication

#### `neo4j.authentication.type`

Selects the database authentication method.

* **Type**: `STRING`
* **Default**: `BASIC`
* **Importance**: High
* **Valid Values / Notes**: `NONE`, `BASIC`, `KERBEROS`, `BEARER`, or `CUSTOM`, case-sensitive. Required credentials for the selected method must not be empty.

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

Configures the BASIC username.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required and non-empty for BASIC.

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

Configures the BASIC password.

* **Type**: `PASSWORD`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required and non-empty for BASIC. Do not write passwords to logs or shared configuration files.

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

Configures the BASIC authentication realm.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: Low
* **Valid Values / Notes**: Optional; specify according to the database authentication settings.

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

Configures the Kerberos ticket.

* **Type**: `PASSWORD`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required and non-empty for KERBEROS. Manage it as a sensitive credential.

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

Configures the Bearer Token.

* **Type**: `PASSWORD`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required and non-empty for BEARER. Manage it as a sensitive credential.

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

Configures the custom authentication scheme.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required and non-empty for CUSTOM; must match the server-side authentication scheme.

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

Configures the custom authentication principal.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required and non-empty for CUSTOM.

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

Configures the custom authentication credentials.

* **Type**: `PASSWORD`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required and non-empty for CUSTOM. Manage them as sensitive credentials.

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

Configures the custom authentication realm.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Optional for CUSTOM; must match the server-side configuration.

### Encryption and Certificates

#### `neo4j.security.encrypted`

Controls encryption for connections using plain URIs.

* **Type**: `STRING`
* **Default**: `false`
* **Importance**: Low
* **Valid Values / Notes**: Accepts only `true` and `false`. When a URI uses `+s` or `+ssc`, the URI controls encryption behavior; this setting does not override it.

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

Selects the certificate trust strategy for explicitly encrypted connections.

* **Type**: `STRING`
* **Default**: `TRUST_SYSTEM_CA_SIGNED_CERTIFICATES`
* **Importance**: Low
* **Valid Values / Notes**: `TRUST_ALL_CERTIFICATES`, `TRUST_SYSTEM_CA_SIGNED_CERTIFICATES`, or `TRUST_CUSTOM_CA_SIGNED_CERTIFICATES`. Applies to explicit encryption with plain URIs; `+s` and `+ssc` use their own trust behavior. Trusting all certificates weakens identity verification.

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

Controls hostname verification for explicit certificate trust strategies.

* **Type**: `STRING`
* **Default**: `true`
* **Importance**: Low
* **Valid Values / Notes**: Accepts only `true` and `false`. Takes effect when an explicit trust strategy is used and does not override the URI's built-in strategy.

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

Specifies custom CA certificate files.

* **Type**: `LIST`
* **Default**: Empty list
* **Importance**: Low
* **Valid Values / Notes**: When using the custom CA trust strategy, provide absolute file paths readable inside the Worker, separated by commas. Do not rely solely on UI hints to determine whether certificates are required.

### Connection Pool and Retries

#### `neo4j.connection-timeout`

Sets the timeout for establishing a connection.

* **Type**: `STRING`
* **Default**: `30s`
* **Importance**: Low
* **Valid Values / Notes**: Nonnegative integers with lowercase units `ms`, `s`, `m`, `h`, or `d`; combinations such as `1m30s` are supported. All duration settings should use this complete format.

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

Limits the driver connection pool size.

* **Type**: `INT`
* **Default**: `100`
* **Importance**: Low
* **Valid Values / Notes**: At least `1`. This is not the number of reading Tasks or safe source shards.

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

Sets the timeout for acquiring a connection from the pool.

* **Type**: `STRING`
* **Default**: `1m`
* **Importance**: Low
* **Valid Values / Notes**: Use a nonnegative duration with lowercase units, such as `30s`.

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

Sets the maximum connection lifetime.

* **Type**: `STRING`
* **Default**: `1h`
* **Importance**: Low
* **Valid Values / Notes**: Use a nonnegative duration with lowercase units.

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

Sets how long a connection must be idle before it is checked prior to reuse.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: Low
* **Valid Values / Notes**: An empty value disables this check. A non-empty value must be a nonnegative duration with lowercase units. The default is not the numeric value `-1`.

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

Sets the maximum retry duration for driver-managed transactions.

* **Type**: `STRING`
* **Default**: `30s`
* **Importance**: Low
* **Valid Values / Notes**: Use a nonnegative duration with lowercase units. This does not mean unlimited automatic Task recovery and does not replace troubleshooting and restarts.

### Read Strategy and Starting Position

#### `neo4j.source-strategy`

Selects the read mode.

* **Type**: `STRING`
* **Default**: `QUERY`
* **Importance**: High
* **Valid Values / Notes**: `QUERY` or `CDC`. QUERY requires a query and a Topic. CDC requires at least one Topic Pattern and does not accept `RAW_JSON_STRING`.

#### `neo4j.start-from`

Selects the starting position when no reusable Offset is available.

* **Type**: `STRING`
* **Default**: `NOW`
* **Importance**: High
* **Valid Values / Notes**: `EARLIEST`, `NOW`, or `USER_PROVIDED`. For QUERY, EARLIEST is `-1` and NOW is the local machine's current epoch time in milliseconds; sequence-number cursors should not use NOW. For CDC, these use the earliest retained ID, the current ID, or a supplied ID, respectively. An available Offset takes precedence for recovery.

#### `neo4j.start-from.value`

Provides a custom starting cursor.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required for USER\_PROVIDED. For QUERY, use a parseable signed Long string; for CDC, use a change ID that remains valid for that database. A query with a strictly-greater-than condition excludes the record at the starting cursor.

#### `neo4j.ignore-stored-offset`

Controls whether stored Offsets are ignored.

* **Type**: `STRING`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: Accepts only `true` and `false`. true forces position selection according to start-from and may replay or skip data. It is reapplied on every restart and is unsuitable as a routine recovery setting.

### Query and Cursor

#### `neo4j.query`

Configures the Cypher query to execute.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required for QUERY. Receives the current cursor through `$lastCheck`. The query must perform its own filtering, sort in ascending order, and return a Long cursor; the Connector does not automatically add WHERE or ORDER BY. Changing the query text changes the source Offset identity.

#### `neo4j.query.streaming-property`

Specifies the result-row field used to advance the cursor.

* **Type**: `STRING`
* **Default**: `timestamp`
* **Importance**: Low
* **Valid Values / Notes**: A non-blank field name; the field in the results must be a Long. Changing the field prevents reuse of existing Offsets. Cursors must be unique and results must be returned in ascending cursor order; sorting alone does not prevent strictly-greater-than pagination from missing records with equal cursors.

#### `neo4j.query.topic`

Specifies the single target Topic for query results.

* **Type**: `STRING`
* **Default**: Empty string
* **Importance**: High
* **Valid Values / Notes**: Required for QUERY. Configuration validation does not check whether the Topic exists. Changing the Topic does not automatically reset the source Offset.

#### `neo4j.query.timeout`

Sets the database transaction timeout.

* **Type**: `STRING`
* **Default**: `0s`
* **Importance**: Low
* **Valid Values / Notes**: A positive duration overrides the transaction timeout; 0 does not override database settings. Shared transaction configuration can also affect CDC transactions. This is not a hard deadline for poll-duration.

#### `neo4j.query.poll-interval`

Sets the wait interval before querying again when QUERY has no data.

* **Type**: `STRING`
* **Default**: `1s`
* **Importance**: Medium
* **Valid Values / Notes**: A nonnegative duration. Shorter intervals increase empty-query load and do not change the Offset commit frequency.

#### `neo4j.query.poll-duration`

Sets the loop time budget for one QUERY poll.

* **Type**: `STRING`
* **Default**: `5s`
* **Importance**: Medium
* **Valid Values / Notes**: Use a positive duration; 0 provides no read time budget. The budget is checked only between operations and does not forcibly interrupt blocking queries or waits.

#### `neo4j.query.force-maps-as-struct`

Controls whether Maps in QUERY results are converted to Structs.

* **Type**: `BOOLEAN`
* **Default**: `true`
* **Importance**: Low
* **Valid Values / Notes**: `true` or `false`. Applies only to QUERY and must match the data structure expected by consumers.

### Batching and Payload Format

#### `neo4j.batch-size`

Limits the number of items read per batch.

* **Type**: `INT`
* **Default**: `1000`
* **Importance**: Medium
* **Valid Values / Notes**: At least `1`. Limits result rows for QUERY and change events for CDC; CDC routing expansion may produce more messages. This is not a byte limit. Monitor memory and downstream load when increasing it.

#### `neo4j.payload-mode`

Selects the payload representation for graph data.

* **Type**: `STRING`
* **Default**: `EXTENDED`
* **Importance**: Medium
* **Valid Values / Notes**: `EXTENDED`, `COMPACT`, or `RAW_JSON_STRING`. EXTENDED uses type wrappers; COMPACT is more compact but requires attention to type changes. RAW\_JSON\_STRING is for QUERY only. Payload mode does not replace a Converter or automatically provide Schema Registry or Schema evolution guarantees.

### CDC Polling and Read Options

#### `neo4j.cdc.use-leader`

Configures the CDC Leader read option.

* **Type**: `BOOLEAN`
* **Default**: `false`
* **Importance**: Medium
* **Valid Values / Notes**: `true` or `false`. Do not treat this option as a guarantee that CDC is forced to route to the Leader. If reads must use a specific node, confirm actual connection behavior together with deployment routing.

#### `neo4j.cdc.poll-interval`

Sets the wait interval when CDC has no data.

* **Type**: `STRING`
* **Default**: `1s`
* **Importance**: Medium
* **Valid Values / Notes**: A nonnegative duration; balance latency requirements against database query load.

#### `neo4j.cdc.poll-duration`

Sets the loop time budget for one CDC poll.

* **Type**: `STRING`
* **Default**: `5s`
* **Importance**: Medium
* **Valid Values / Notes**: Use a positive duration. The budget is checked only between operations and is not a hard timeout for database calls.

### CDC Topic Routing and Event Representation

#### `neo4j.cdc.topic.<topic>.patterns`

Sets node or relationship selection Patterns for the target Topic.

* **Type**: `STRING`
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid Values / Notes**: A dynamic CDC setting; provide at least one actual Pattern. For example, `(:Event)` matches Event nodes. Topic names support letters, digits, dots, underscores, and hyphens. Do not mix this form with the indexed form for the same Topic. Pattern projections change the output properties.

#### `neo4j.cdc.topic.<topic>.patterns.<index>.pattern`

Configures a single CDC selection Pattern by index.

* **Type**: `STRING`
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid Values / Notes**: Indices start at `0` and must be consecutive, with one Pattern per index. Do not mix this form with the non-indexed form for the same Topic.

#### `neo4j.cdc.topic.<topic>.patterns.<index>.operation`

Filters events at this index by change operation.

* **Type**: `STRING`
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid Values / Notes**: `create`, `update`, or `delete`; converted to lowercase at runtime. A Pattern must exist at the same index. Omitting this setting leaves the operation filter unset.

#### `neo4j.cdc.topic.<topic>.patterns.<index>.changesTo`

Filters events by properties that have changed.

* **Type**: `STRING`
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid Values / Notes**: Comma-separated property names, with leading and trailing whitespace removed. A Pattern must exist at the same index.

#### `neo4j.cdc.topic.<topic>.patterns.<index>.metadata.<metadata-key>`

Filters changes at this index by event metadata.

* **Type**: `STRING`
* **Default**: No ConfigDef default
* **Importance**: Not declared
* **Valid Values / Notes**: metadata-key supports `authenticatedUser`, `executingUser`, and `txMetadata.<key>`. Key names support letters, digits, dots, underscores, and hyphens. A Pattern must exist at the same index.

#### `neo4j.cdc.topic.<topic>.key-strategy`

Selects how CDC message Keys are constructed.

* **Type**: `STRING`
* **Default**: No ConfigDef default; uses `WHOLE_VALUE` at runtime when unset
* **Importance**: Not declared
* **Valid Values / Notes**: `SKIP`, `ELEMENT_ID`, `ENTITY_KEYS`, or `WHOLE_VALUE`, case-sensitive. ENTITY\_KEYS outputs null when the entity has no keys; do not assume that all events have a business key.

#### `neo4j.cdc.topic.<topic>.value-strategy`

Selects how CDC message Values are constructed.

* **Type**: `STRING`
* **Default**: No ConfigDef default; uses `CHANGE_EVENT` at runtime when unset
* **Importance**: Not declared
* **Valid Values / Notes**: `CHANGE_EVENT` or `ENTITY_EVENT`, case-sensitive. The former retains the change-event representation; the latter uses an entity-event representation.

### CDC Metric Collection Options

#### `neo4j.cdc.metric.last-db-tx-id.enabled`

Controls whether the latest database transaction ID metric is collected.

* **Type**: `STRING`
* **Default**: `false`
* **Importance**: Low
* **Valid Values / Notes**: Accepts only `true` and `false`. Enabling it adds the corresponding refresh queries.

#### `neo4j.cdc.metric.last-db-tx-id.refresh-interval`

Sets the refresh interval for the latest transaction ID metric.

* **Type**: `STRING`
* **Default**: `30s`
* **Importance**: Low
* **Valid Values / Notes**: A nonnegative duration; meaningful only when this metric is enabled.

### Connector Execution and Serialization

#### `connector.class`

Specifies the Source Connector implementation class.

* **Type**: `STRING`
* **Default**: None, required
* **Importance**: High
* **Valid Values / Notes**: Use `org.neo4j.connectors.kafka.source.Neo4jConnector`.

#### `tasks.max`

Sets the maximum number of Tasks.

* **Type**: `INT`
* **Default**: `1`
* **Importance**: High
* **Valid Values / Notes**: At least `1`. Keep one Task for a single configured data flow. Multiple Tasks independently read the same source data and share the logical Offset identity; they do not implement safe source sharding.

#### `key.converter`

Specifies the Converter for message Key serialization.

* **Type**: `CLASS`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Leave unset to inherit the Worker configuration. When set, use an instantiable Converter class such as `org.apache.kafka.connect.json.JsonConverter` and configure serialization options as required by that Converter.

#### `value.converter`

Specifies the Converter for message Value serialization.

* **Type**: `CLASS`
* **Default**: `null`
* **Importance**: Low
* **Valid Values / Notes**: Leave unset to inherit the Worker configuration. When set, use an instantiable Converter class. It must handle the data types produced by the selected payload-mode.

## Best Practices

### Stream Graph Creates, Updates, and Deletes

**Business Scenario**: An existing database needs to send subsequent graph changes to downstream applications. Use CDC rather than querying current node state, especially when hard deletes must be transmitted. Enable CDC before starting; plan a separate baseline export for existing entities rather than relying on CDC to populate it automatically.

**Configuration Example**: Reuse the connection and authentication settings from Quick Start, remove `neo4j.query` and `neo4j.query.topic`, override `neo4j.start-from`, and add the following settings. Replace `<cdc-topic>` in the dynamic key with the actual Topic name.

```properties theme={null}
neo4j.source-strategy=CDC
neo4j.start-from=NOW
neo4j.cdc.topic.<cdc-topic>.patterns=(:Event)
```

**Key Notes**: For a new Connector without a reusable Offset, NOW reads Event node changes after the current CDC position; subsequent operation uses stored Offsets. Design the handoff between the baseline and change stream within the source retention period to avoid gaps in changes during export. This Pattern does not restrict operation types, so matching creates, updates, and deletes can all be sent; CHANGE\_EVENT is used by default. This configuration does not automatically enable exactly-once. Under ordinary operation, consumers should handle replay after failure recovery.

### Replay Query Records from a Specified Position After Maintenance

**Business Scenario**: A QUERY data flow needs to resend data after a known cursor, for example to reprocess records after fixing downstream processing issues. The required events remain available at the source, and consumers are ready for deduplication or idempotent processing.

**Configuration Example**: Override the starting position in the Quick Start QUERY configuration and add the following settings. `<last-processed-sequence>` is the last sequence number before the range to replay; it is a numeric source position, not a Kafka Offset.

```properties theme={null}
neo4j.start-from=USER_PROVIDED
neo4j.start-from.value=<last-processed-sequence>
neo4j.ignore-stored-offset=true
```

**Key Notes**: Pause the Connector before changing its configuration. The query's strictly-greater-than condition reads only records after the specified sequence number. Replay may produce duplicate messages and can read only records that still exist; it cannot recover hard-deleted data. After confirming that messages have been sent and the new Offset has been persisted, remove `neo4j.ignore-stored-offset` and `neo4j.start-from.value`, restore `neo4j.start-from` to `EARLIEST`, and resume routine operation. Keep the database, query text, and cursor field unchanged before resuming to avoid selecting a new starting point.

## Monitoring

### What to Monitor

Monitor Kafka Connect cluster health, Connector and Task states, throughput, processing latency, Offset commits, errors, and retries, along with Worker JVM heap memory, GC, and thread states. If the status is healthy but no messages arrive, use source data changes and consumption results to determine whether the flow is actually idle. Monitor DLQ activity only when the corresponding error handling is enabled in the deployment.

### Import the Grafana Dashboard

Download the [AutoMQ Connect Cluster Dashboard](https://automq-download-center.oss-cn-hangzhou.aliyuncs.com/connect-dashboard/automq-connect-cluster-dashboard.json), confirm that Kafka Connect metrics are connected to a Prometheus-compatible data source and that cluster, Worker, Connector, and other labels match the dashboard queries, then import the JSON into Grafana and select the appropriate data source.

## Limitations

* QUERY stores only a single Long cursor and provides no secondary pagination position for equal cursor values; strictly-greater-than conditions may miss records when equal cursor values occur at batch boundaries.
* QUERY does not automatically generate filtering or sorting conditions and cannot automatically capture hard deletes.
* CDC EARLIEST reads only retained change history; it does not export all existing entities.
* CDC change IDs depend on database retention and history. Backup restoration, snapshot restoration, and resuming Aura after a pause may invalidate existing IDs; do not rely on automatic fallback.
* CDC cannot mix list and indexed Patterns for the same Topic; indices must be consecutive starting at zero.
* Overlapping CDC selectors may produce multiple messages for the same change; no cross-selector deduplication is performed.
* Multiple Tasks do not provide source sharding; increasing the Task count may cause duplicate reads and Offset interference.
* QUERY does not support Source exactly-once. CDC's support statement still requires distributed Workers to enable the capability, Brokers to support transactions with appropriate authorization, and consumers to read with transaction isolation; it must not be treated as an automatic end-to-end guarantee.
* Global ordering across Kafka partitions, Topics, or multiple Tasks is not guaranteed. Sequence numbers for CDC events in the same transaction also do not guarantee the order in which the original operations were executed.
* CDC does not accept the `RAW_JSON_STRING` payload mode.

## FAQ

### What Should I Do If a Task Is Running Normally but No New Messages Arrive?

For QUERY, first check the target database, Topic, and query results. Confirm that the returned fields include the configured Long cursor and that new records have cursor values greater than the stored position. The default NOW uses the current epoch time in milliseconds; if the business field is an ordinary sequence number, change it to EARLIEST on the first start. When an Offset already exists, changing only start-from does not reset the position. For CDC, check whether CDC is enabled on the database, the account has the necessary permissions, and the actual Pattern matches. NOW does not resend entities that existed before startup. Before replaying, confirm source data retention and explicitly select a starting point.

### What Should I Do If Records Are Missing After Query Reads?

Check whether the query returns results in ascending cursor order, whether cursors are unique, and whether later writes use cursor values that have already been passed. The Connector advances the cursor using the last row in the batch; equal timestamps or unordered results may create gaps. Use unique monotonically increasing sequence numbers or separate event records. Missing records that are still retained can be replayed from an earlier position once downstream deduplication is ready. Adding ORDER BY alone cannot resolve pagination boundaries with equal cursor values.

### What Should I Do If Duplicate Messages Appear After a Restart?

In ordinary Source operation, a failure after messages are published but before the Offset is persisted may cause records to be read again during recovery. Check whether ignore-stored-offset=true has been left enabled, whether the database, query text, or cursor field has changed, and whether multiple identical reading Tasks have been deployed. Keep one Task, maintain a stable source identity, and restore default Offset reuse; consumers should deduplicate using business identifiers and event versions. CDC exactly-once requires the platform to confirm the transaction conditions for Workers, Brokers, and consumers. Setting source-strategy alone does not provide this guarantee.

### What Should I Do If CDC Reports an Invalid Change ID During Recovery?

Check the downtime against the CDC retention period and whether the database was restored or Aura was resumed after a pause. An invalidated position can no longer be used for continuous reading. First determine the missing range and a plan to rebuild the baseline, then select a valid USER\_PROVIDED ID, the earliest retained position, or the current position. Forcing stored Offsets to be ignored changes the read range; it cannot repair history that has already expired.

### Why Is Old Data Not Backfilled After Changing the Topic or Expanding CDC Selection?

The Topic, connection URI, and CDC selectors are not part of the corresponding source Offset identity. The existing Connector may continue using its previous position and will not automatically backfill data because the output scope changed. First determine the read position to retain and the range that needs backfilling. If historical replay is required, confirm retention, deduplication, and baseline handoff before explicitly adjusting the starting point. Do not assume that changing the Topic causes a fresh export.

### What Should I Do If a Task Fails Because of Query Field Types or Authentication Errors?

Check the full exception logs without exposing credentials. The QUERY cursor must exist and be a Long; correct the Cypher return fields and source types before resuming. For authentication failures, verify the selected authentication type and its required credentials. For certificate failures, check URI encryption and certificate paths inside the Worker. The maximum transaction retry duration is not an automatic recovery strategy for all Task errors. After correcting the configuration, restart failed Tasks through Connector management operations.