Source Configuration Settings

org.neo4j.connectors.kafka.source.Neo4jConnector

A compatible Kafka Connect converter.

Could be one of;

Schema Registry URL for message keys. Required when a schema registry is required by the configured key.converter.

Should be set to true when key.converter is io.confluent.connect.protobuf.ProtobufConverter.

A compatible Kafka Connect converter.

Could be one of;

Schema Registry URL for message values. Required when a schema registry is required by the configured value.converter.

Should be set to true when value.converter is io.confluent.connect.protobuf.ProtobufConverter.

Neo4j URI to connect to. Multiple URIs can be specified separated by ,.

Neo4j database name to connect to. Recommended to be specified explicitly.

Authentication provider to use. One of none, basic, kerberos, bearer, custom, oidc, or the name of your own provider. The value is case-insensitive.

Default: basic

Username to authenticate with. Required when neo4j.authentication.type is basic.

Password to authenticate with. Required when neo4j.authentication.type is basic.

Authentication realm to authenticate with, leave empty for default.

Kerberos ticket to establish connection with. Required when neo4j.authentication.type is kerberos.

Bearer token to establish connection with. Required when neo4j.authentication.type is bearer.

Custom authentication scheme to establish connection with. Required when neo4j.authentication.type is custom.

Custom principal to establish connection with. Required when neo4j.authentication.type is custom.

Custom credential to establish connection with. Required when neo4j.authentication.type is custom.

Custom authentication realm to authenticate with, set as required by your custom authentication provider.

Issuer URL. The token endpoint is read from {issuer}/.well-known/openid-configuration. One of issuer, discoveryUrl or tokenEndpoint is required when neo4j.authentication.type is oidc.

Discovery document URL, given directly. The document must contain issuer and token_endpoint.

Default: derived from issuer

Issuer that the discovery document must contain. Set it only when the identity provider is reached at one address and advertises another, for example behind a reverse proxy. Must match the iss claim of the tokens.

Default: value of issuer

Token endpoint URL, given directly. Skips discovery.

Default: from the discovery document

OAuth 2.0 grant used to obtain tokens. One of client_credentials, password, refresh_token, jwt_bearer, token_exchange.

Default: client_credentials

Space-separated scopes.

Value of the audience request parameter.

Value of the resource request parameter.

Additional request parameter <name>, sent as is. Overrides any other parameter with the same name.

Username. Required when neo4j.authentication.oidc.grantType is password.

Password. Required when neo4j.authentication.oidc.grantType is password.

Initial refresh token. The connector uses the new refresh token when the identity provider rotates it. Required when neo4j.authentication.oidc.grantType is refresh_token.

File that holds the subject token. The file is read on every token request. Required when neo4j.authentication.oidc.grantType is token_exchange.

Subject token type, for the token_exchange grant.

Default: urn:ietf:params:oauth:token-type:jwt

Client identifier. Required unless neo4j.authentication.oidc.clientAuthMethod is none.

Client secret.

File that holds the client secret, instead of clientSecret. The file is read on every token request, so a rotated secret is picked up.

How the client authenticates to the token endpoint. One of client_secret_basic, client_secret_post, private_key_jwt, none. Cannot be none with the client_credentials grant.

Default: client_secret_basic

PKCS#8 private key file, PEM or DER, for private_key_jwt and the jwt_bearer grant.

Key identifier, the kid header of the signed JWT.

Signing algorithm. One of RS256, RS384, RS512, PS256, PS384, PS512, ES256, ES384, ES512.

Default: RS256

Claim <name> of the signed JWT. iss, sub, aud, iat, exp and jti are set automatically unless set here. Integer and boolean values are sent as such.

Validity of the signed JWT, in seconds.

Default: 300

Response field that holds the token. One of access_token, id_token.

Default: access_token

Number of seconds before expiry at which the token is treated as expired.

Default: 30

Token lifetime, in seconds, when the response has no expires_in and the token is not a JWT.

Default: 300

Connection timeout to the identity provider, in milliseconds.

Default: 10000

Request timeout to the identity provider, in milliseconds.

Default: 30000

Number of retries on timeout and on HTTP 408, 429 and 5xx responses. The connector does not retry invalid_client and invalid_grant errors.

Default: 3

Initial delay between retries, in milliseconds. The delay doubles on every retry, up to 30 seconds. A Retry-After response header takes precedence.

Default: 500

Trust store for TLS connections to the identity provider, for example for a private certificate authority.

Default: JDK trust store

Password of neo4j.authentication.oidc.trustStoreFile.

Type of neo4j.authentication.oidc.trustStoreFile.

Default: PKCS12

HTTP proxy host for connections to the identity provider. Set together with proxyPort.

Default: JDK setting

HTTP proxy port for connections to the identity provider. Set together with proxyHost.

Default: JDK setting

Set of default values for a well-known identity provider. Settings you configure override the profile values. One of google, okta, entra.

TCP connection timeout (valid units are: ms, s, m, h and d; default unit is s).

Default: 30s (Driver default)

Maximum number of connections to keep in the connection pool.

Default: 100 (Driver default)

Maximum duration to wait for acquiring a connection from the connection pool (valid units are: ms, s, m, h and d; default unit is s).

Default: 60s (Driver default)

Duration after which idle connections are tested for liveness (valid units are: ms, s, m, h and d; default unit is s).

Default: no test (Driver default)

Duration after which a connection is dropped from the connection pool (valid units are: `ms, s, m, h and d; default unit is s).

Default: 1h (Driver default)

Maximum duration to retry a transaction.

Default: 30s

Whether encryption is enabled. Only applicable when bolt or neo4j schemes are used inside neo4j.uri. One of true, false

Default: false.

Trust strategy to use for TLS connections. One of TRUST_ALL_CERTIFICATES, TRUST_SYSTEM_CA_SIGNED_CERTIFICATES, TRUST_CUSTOM_CA_SIGNED_CERTIFICATES Required when neo4j.security.encrypted is true.

Default: TRUST_SYSTEM_CA_SIGNED_CERTIFICATES

List of files that contain X509 certificates of CAs to trust. Required when neo4j.security.trust-strategy is TRUST_CUSTOM_CA_SIGNED_CERTIFICATES.

Whether hostname verification is enabled during TLS handshake. One of true, false.

Default: true

Source strategy for this connector. One of CDC, QUERY.

A time anchor to start streaming from. One of EARLIEST, NOW, USER_PROVIDED. Only used on initial run of the connector instance, and ignored when there is already a stored offset in Kafka Connect.

Default: NOW

Custom value to use as a starting offset. Required when neo4j.start-from is set to USER_PROVIDED.

Whether to ignore any offset value retrieved from the offset storage saved by a previous run. One of true, false.

Default: false

Maximum number of change events to publish for each poll cycle.

Default: 1000

Defines the structure of change messages. One of COMPACT, EXTENDED, RAW_JSON_STRING. COMPACT provides simpler messages but faces schema compatibility issues if property types change. EXTENDED includes type information to avoid such issues. RAW_JSON_STRING produces a raw JSON string representation of the data.

Default: EXTENDED. Introduced in 5.1.5

Serialisation strategy for CDC topic key. One of SKIP, ELEMENT_ID, ENTITY_KEYS, WHOLE_VALUE.

Default: WHOLE_VALUE

For guidance on choosing a strategy, see Best practices for Source connector.

Serialisation strategy for CDC topic value. One of CHANGE_EVENT, ENTITY_EVENT.

Default: CHANGE_EVENT

Note: If you’re using a sink connector with one of the CDC strategies, this setting must be configured as CHANGE_EVENT.

Maximum amount of time Kafka Connect poll request will wait for a change to be received from the database (valid units are: ms, s, m, h and d; default unit is s)

Default: 5s

The interval in which the database will be polled for changes during neo4j.cdc.poll-duration (valid units are: ms, s, m, h and d; default unit is s).

Default: 1s

Comma-separated list of graph patterns, with optional event property filters (prefixed with + or - for inclusion or exclusion respectively).

Example setting: "neo4j.cdc.topic.my-topic.patterns": "(:Person {+name}),(:Person)-[:WORKS_FOR]→(:Company),(:Company {-id})". The change event for Company nodes will not include the id property.

For guidance on choosing between this form and the indexed form below, see Choosing between the list and indexed forms.

Indexed graph pattern, with optional event property filters (prefixed with + or - for inclusion or exclusion respectively).

Example: "neo4j.cdc.topic.my-topic.patterns.0.pattern": "(:Company {+name})". The change event for Company nodes will either always include the name property (in CDC FULL enrichment mode) or when the name property has changed (in CDC DIFF enrichment mode).

One of CREATE, UPDATE, DELETE. Operation for which we want to receive events for the corresponding indexed graph pattern.

Example: "neo4j.cdc.topic.my-topic.patterns.0.operation": "DELETE"

Comma-separated list of property names.

Example: "neo4j.cdc.topic.my-topic.patterns.0.changesTo": "name, age" (see how selectors are combined)

User who was authenticated while performing the change.

Example: "neo4j.cdc.topic.my-topic.patterns.0.metadata.authenticatedUser": "userA"

User who was executing the transaction while performing the change. Usually will be same with authenticatedUser, but might change if user impersonation is used.

Example: "neo4j.cdc.topic.my-topic.patterns.0.metadata.executingUser": "userB"

Key-value transaction metadata selector. The key-value pair is matched against the actual transaction metadata (with the key stripped of the aforementioned prefix).

Example: "neo4j.cdc.topic.my-topic.patterns.0.metadata.txMetadata.app": "neo4j-browser"

Whether to enable the last_db_tx_id metric.

Default: false Introduced in 5.3.0

The refresh interval of the last_db_tx_id metric (valid units are: ms, s, m, h and d; default unit is s).

Default: 30s Introduced in 5.3.0

Whether to always run CDC queries against the leader instance of the database, instead of any cluster member.

For guidance on choosing between the two, see Decide whether to pin CDC queries to the leader.

Default: false Introduced in 5.3.0

Kafka topic to publish change events gathered through provided query.

Cypher query to use for gathering changes. Requires both neo4j.query.streaming-property to be in the result set, and $lastCheck query parameter used for tracking changes.

Property name that is both present in the result set of the specified query and used as a filter to query changes from a previous value.

Default: timestamp

Maximum amount of time Kafka Connect poll request will wait for a change to be received from the database (valid units are: ms, s, m, h and d; default unit is s).

Default: 5s

The interval in which the database will be polled for changes during neo4j.query.poll-duration (valid units are: ms, s, m, h and d; default unit is s).

Default: 1s

Maximum amount of time query is allowed to run (valid units are: ms, s, m, h and d; default unit is s).

Default: 0s

Schema type used for map values returned by the query. One of STRUCT, MAP, LEGACY.

The message itself is always a STRUCT with one field per returned column. Only STRUCT is allowed with the RAW_JSON_STRING payload mode. The Protobuf converter cannot serialize a list of maps with MAP or LEGACY. Replaces neo4j.query.force-maps-as-struct, see Migrate from 5.x to 6.0.

Default: STRUCT