Kafka Connector | Memority Documentation

Introduction

This page describes the configuration and usage of the Kafka Connector. The Kafka Connector enables integration with Apache Kafka brokers. It supports two complementary use cases:

Outbound provisioning: producing Kafka messages in reaction to IM object events, such a IM identities creation or update. A provisioning operation is implemented by a Groovy rule that builds a Kafka message
Live Inbound synchronization: consuming Kafka messages in real time from one or more topics and synchronizing them as Memority IM objects. Each consumed message is mapped to an IM object by a Groovy rule.

A single KafkaConnectorDefinition may be configured for outbound only, inbound only, or both at the same time.

Limitations

Outbound provisioning: the GET_OBJECT and SEARCH_OBJECTS Connector operations are not supported. A Kafka broker is not a queryable datastore: it cannot be searched or fetched. As a consequence, the provisioning service always considers that a Kafka account must be created, so in practice CREATE_OBJECT is the operation that gets invoked. Other write operations, i.e. PATCH_OBJECT and DELETE_OBJECT may be invoked in reaction to an IM object event such as:
- IM object deleted
- IM object disabled
- IM object loses the role triggering the Kafka provisioning
- this behavior is configured as usual through the ApplicationObjectActivationDefinition section of the Application
Account Discovery is not supported for the same reason.

Configuration

You can access the Kafka Connector configuration:

by clicking on "Synchronization" → "Kafka Connectors" (not implemented yet)
by clicking on "System" → "Configurations" → "Synchronization Service" and performing an import/export.

The REST API URL to manage Kafka Connector definitions is: /{tenant}/api/sync/conf/kafka-connector-definitions.

General Configuration Principles

A Connector is usually configured using simple key/value properties. The Kafka Connector configuration is however far more complex; the key/value model is not adapted to its complexity.

The principles are thus:

Configure the Kafka Connector using a specialized KafkaConnectorDefinition, which is a regular configuration aggregate.
Configure as usual a ConnectorDefinition (of class com.memority.domino.connector.kafka.KafkaConnector) which holds a single property kafkaConnectorDefinitionId whose value references the id of the above KafkaConnectorDefinition.

KafkaConnectorDefinition Properties

The following table lists the top-level configuration properties of KafkaConnectorDefinition.

Property Name	Type	Mandatory	Description
bootstrapServers	`String`	YES	The Kafka bootstrap servers, as the comma-separated list of `host:port` expected by the Kafka `bootstrap.servers` property, for instance `broker1.acme.com:9094,broker2.acme.com:9094`. May be `settings:some.setting.key` that references a Setting of type String, which is the recommended way to point each environment at its own brokers. Listing more than one server is recommended for high-availability purpose.
producerDefinition	`ProducerDefinition`	NO	Pertains to the outbound provisioning case: how Kafka messages are reliably sent to topics. Can be left null if only inbound synchronization is needed. See Producer Definition
consumerDefinition	`ConsumerDefinition`	NO	Pertains to the inbound synchronization case: how messages are consumed from topics in order to be synchronized as IM objects. Can be left null if only outbound provisioning is needed. See Consumer Definition
transportDefinition	`TransportDefinition`	NO	How the connection to the Kafka broker is encrypted. Defaults apply if not configured, i.e. TLS is enabled and the brokers' certificates are validated against the JVM default truststore. See Transport Definition
authenticationDefinition	`AuthenticationDefinition`	NO	How to authenticate against the Kafka broker. If not configured, the Connector does not authenticate. Only mTLS (Mutual Transport Layer Security) is supported. See Authentication Definition

Producer Definition

Pertains to the outbound provisioning case only. How reliably messages are sent to the Kafka broker.

Property Name	Type	Mandatory	Description	Default Value
operationDefinitions	`List<OperationDefinition>`	YES	Pertains to the outbound provisioning case: the rules producing the messages sent to Kafka, one per supported provisioning operation. See Operation Definition
defaultTopic	`String`	NO	The topic messages are sent to, unless overridden by `OperationDefinition.topic` or by the topic set via the `KafkaMessageBuilder` in the Groovy script. May be `settings:some.setting.key`.
defaultPartition	`Integer`	NO	The partition messages are sent to, unless overridden at the operation or script level. If no partition is resolved, the partition is automatically assigned by the Kafka broker, according to the message key if any, or randomly.
clientId	`String`	NO	An identifier passed to the broker, useful to track the origin of requests in broker logs and quotas.	Defaults to `memority-<tenant>-<KafkaConnectorDefinition id>`.
acks	`Enum`	NO	The number of broker acknowledgements required before considering a message as sent. Values: `NONE` (no acknowledgement), `LEADER` (leader only), `ALL` (all in-sync replicas, default and safest).	`ALL`
sendTimeout	`Duration`	NO	The maximum time to wait for the Kafka broker acknowledgement of a message. It should be greater than `deliveryTimeout`. Range: 100 ms – 2 minutes.	60 seconds
requestTimeout	`Duration`	NO	The maximum time to wait for a broker response to a single request. Range: 100 ms – 5 minutes.	30 seconds
deliveryTimeout	`Duration`	NO	The overall time budget of a send, retries included. Must be greater than `requestTimeout`. Range: 100 ms – 10 minutes.	45 seconds
maxBlock	`Duration`	NO	The maximum time a send may block waiting for metadata or for buffer space. This is what bounds the very first send when the brokers are unreachable. Range: 100 ms – 30 seconds.	15 seconds
retries	`Integer`	NO	The number of times a failed send is retried. Retries are bounded by `deliveryTimeout`. Range: 0 – 10.	2
retryBackoff	`Duration`	NO	The time to wait before retrying a failed send. Range: 0 ms – 30 seconds.	1 second
maxRequestSizeBytes	`Integer`	NO	The maximum size of a single message. Must be consistent with the broker's own `message.max.bytes` setting. Range: 1 024 – 67 108 864 (64 MiB).	1 048 576 (1 MiB)
compressionType	`Enum`	NO	The compression applied to the messages produced. Values: `NONE`, `GZIP`, `SNAPPY`, `LZ4`, `ZSTD`.	`NONE`
enableIdempotence	`Boolean`	NO	Whether the producer guarantees that retries won't produce duplicate messages. Requires `acks = ALL` and `retries > 0`.	`true`

Operation Definition

Pertains to the outbound provisioning case only.

Configuration of a single Connector operation, i.e. the Groovy rule producing the Kafka message corresponding to that operation.

Only the following operations are supported: CREATE_OBJECT, PATCH_OBJECT, DELETE_OBJECT. The GET_OBJECT, SEARCH_OBJECTS and DISCOVER_OBJECTS operations are not supported because a Kafka broker cannot be queried.

Property Name	Type	Mandatory	Description
operation	`ConnectorOperation`	YES	The name of the Connector operation: `CREATE_OBJECT`, `PATCH_OBJECT` or `DELETE_OBJECT`.
action	`RuleDefinition`	YES	The Groovy script producing the Kafka message, which must return a `KafkaMessageBuilder`, typically using `KAFKA.message { ... }`. More on that below.
topic	`String`	NO	The topic the messages of this operation are sent to, overriding `ProducerDefinition.defaultTopic`. It is itself overridden by the topic set via the `KafkaMessageBuilder` in the Groovy script, if any.
partition	`Integer`	NO	The partition the messages of this operation are sent to, overriding `ProducerDefinition.defaultPartition`. It is itself overridden by the partition set via the `KafkaMessageBuilder` in the Groovy script, if any.

Consumer Definition

Pertains to the inbound synchronization case only. How messages are consumed from Kafka topics in order to be synchronized with IM objects. This is the inbound counterpart of the outbound ProducerDefinition.

Consumption is driven by a SynchronizationTaskDefinition,which namely provides the object class, the shadow kind and the attribute mappings, while this definition provides the topics to read from and the rule turning a Kafka message into an IM object. Unlike a regular import or reconciliation task, such a task is never scheduled: it is live and runs for as long as the tenant is up.

Property Name	Type	Mandatory	Description	Default Value
topics	`String`	YES	The topics to consume from, as a comma-separated list, for instance `employees,partners`. May be `settings:some.setting.key`.
messageToAccountConverter	`RuleDefinition`	YES	The Groovy rule turning a consumed Kafka message into an `Account`. The incoming Kafka message is bound to the rule as the `KAFKA_MESSAGE` variable (of type `KafkaMessage`), and the rule must return an `Account`. See Inbound Groovy Context
objectUniqueIdentifier	`String`	YES	The name of the Account’s attribute (the Account returned by `messageToAccountConverter)` holding the identifier of the account on the remote application. An Account that does not carry that attribute is reported as an error and skipped.
enabled	`Boolean`	NO	Whether messages are actually consumed. Setting it to `false` stops the consumption without having to drop the whole configuration, which is handy to temporarily relieve a struggling or overloaded consumption	`true`
autoOffsetReset	`Enum`	NO	Where to start reading a partition that has no committed offset yet. Values: `LATEST` (the safe default: ignores topic history), `EARLIEST` (replays the whole topic history), `NONE` (fail instead of silently replaying or skipping).	`LATEST`
concurrency	`Integer`	NO	The number of threads consuming in parallel. A partition is consumed by a single thread, hence configuring more threads than the topics have partitions leaves the extra threads idle (not recommended). Range: 1 – 32.	1
maxPollRecords	`Integer`	NO	The maximum number of messages handed over to the synchronization at once. A batch is synchronized as a whole before its offsets are committed, so this bounds how many messages may be synchronized twice should the processing stop mid-batch. Range: 1 – 5 000.	500
maxPollInterval	`Duration`	NO	The maximum time a batch may take to be synchronized before the broker considers the consumer dead and rebalances its partitions. It must exceed the time needed to synchronize `maxPollRecords` messages, IM calls included. Range: 1 second – 2 hours.	30 minutes
pollTimeout	`Duration`	NO	How long a poll waits for messages before returning empty. It only affects how quickly an idle consumer notices it must stop. Range: 100 ms – 1 minute.	5 seconds

Transport Definition

This section defines how the connection to the Kafka brokers is secured at the transport level, i.e. the TLS settings.

Property Name	Type	Mandatory	Description	Default Value
tlsEnabled	`Boolean`	NO	Whether the connection to the brokers is encrypted.	`true`
tlsProtocol	`Enum`	NO	The preferred TLS protocol version. Values: `TLS_1_2`, `TLS_1_3`	`TLS_1_3`
enabledTlsProtocols	`List<TlsProtocol>`	NO	The TLS protocol versions accepted when negotiating with the broker.	`[TLS_1_2, TLS_1_3]`
trustedCertificates	`String`	NO	PEM-encoded certificates trusted when validating the Kafka brokers' certificates. May be `settings:some.setting.key`. If not configured, the JVM default truststore applies, which is enough when the brokers' certificates are issued by a well-known public CA.
bypassTlsValidation	`Boolean`	NO	If `true`, the Kafka brokers' server certificate is not checked, and as a side-effect mTLS client authentication is disabled. Warning: this opens the door to man-in-the-middle attacks and should only be used for tests.	`false`

Authentication Definition

This section defines how the Kafka Connector authenticates against the Kafka broker. Authentication is optional. Only mutual TLS (mTLS) is supported at this time. Additional SASL mechanisms (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI, OAUTHBEARER) may be added in the future.

Mutual TLS Authentication Definition

Authenticate against the Kafka brokers with a client certificate.

Property Name	Type	Mandatory	Description
mechanism	`String`	YES	Must be `mtls`.
certificateChain	`String`	YES	PEM-encoded client certificate presented to the Kafka broker, followed by the certificates of its issuers if the broker needs the whole chain. May be `settings:some.setting.key`.
privateKeyUri	`String`	YES	A URI to the PEM-encoded private key (PKCS#8), typically `settings:some.secret.setting.key` referencing a `SecretSetting`. Must never be configured inline, use a secret Setting indirection.
privateKeyPasswordUri	`String`	NO	A URI to the password protecting the private key, typically `settings:some.secret.setting.key`. Only needed if the private key is encrypted. Must never be configured inline, use a secret Setting indirection.

Writing Groovy Scripts

This section explains how to craft Groovy scripts aimed at:

generating Kafka messages for the outbound provisioning case
reading Kafka messages for the inbound synchronization case

Outbound Provisioning Scripts

3 connector operations may be configured as Groovy scripts:

CREATE_OBJECT: generate a Kafka message for both IM object events: IM object creation and IM object update. As mentioned earlier, since the Kafka Connector has no read capability, “create” and “update” events cannot be distinguished

PATCH_OBJECT and DELETE_OBJECT may be configured in reaction to an IM object event such as:
- IM object deleted
- IM object disabled
- IM object loses the role triggering the Kafka provisioning
- this behavior is configured as usual through the ApplicationObjectActivationDefinition section of the Application
- PATCH_OBJECT is only relevant if a “custom action” Rule has been configured in ApplicationObjectActivationDefinition, because the “custom action” Rule produces an ObjectPatch that is then played via PATCH_OBJECT

Those 3 operations correspond to 3 OperationDefinition items (see above). Each Groovy script must return a KafkaMessageBuilder, which is a fluent builder for a Kafka message.

The KafkaMessageBuilderAPI is detailed below.

Outbound Groovy Contexts

Groovy scripts access external values through contexts (the technical term is "bindings") that are injected upon execution.

Context Variable Name	Java Type	Description	Available For Operation
`KAFKA`	`GroovyKafkaClient`	The Kafka client on which the `message` method is called to build a Kafka message.	All operations
`IDM_OBJECT`	`ApiObject`	The IM object, source of provisioning.	All operations
`ACCOUNT`	`Account`	The object to process on the remote application, whose attributes are the target application's attributes (not IM attributes).	All operations
`ACCOUNT_ID`	`String`	The identifier of the account known by Memority.	PATCH, DELETE
`ACTIVATION_SITUATION`	`ActivationSituation` (enum)	May be null. A non-null value indicates one of the possible situations: `IDM_OBJECT_DELETED`, `IDM_OBJECT_DISABLED`, `APPLICATION_UNASSIGNED`.	PATCH, DELETE
`PATCH`	`Map<String, AttributeDelta>`	The attribute patches to be applied.	PATCH

KafkaMessageBuilder API

The KafkaMessageBuilder exposes the following methods. Examples are provided below.

Method	Description
`topic(String topic)`	Set the topic to which the message will be sent. If not specified, the topic is retrieved from the `OperationDefinition` or `ProducerDefinition`.
`partition(Integer partition)`	Set the partition to which the message will be sent. If not specified, the partition is retrieved from the configuration.
`header(String name, Object value)`	Add a header to the message. If the header is already present, it is replaced. Multivalued headers are not supported.
`key(Object key)`	Set the message key. For the `CREATE_OBJECT` operation, if a key is set, it is used as the account's unique identifier. If not key is set by the Groovy script, then an attribute mapping must be configured for the`__objectId__`attribute, and the resulting mapping value is taken as account identifier.
`value(Object json)`	Set the message payload, JSON-serializable, typically a `Map<String, Object>`.

Connector Operations

Connector Operation	Description	Expected Groovy Script Output
`CREATE_OBJECT`	Produce a Kafka message corresponding to the creation or update of an account. The message key, if set, is used as the account's unique identifier. If not key is set by the Groovy script, then an attribute mapping must be configured for the`__objectId__`attribute, and the resulting mapping value is taken as account identifier.	`KafkaMessageBuilder`
`PATCH_OBJECT`	Produce a Kafka message corresponding to the update of an account for a custom action configured in `ApplicationObjectActivationDefinition`.	`KafkaMessageBuilder`
`DELETE_OBJECT`	Produce a Kafka message corresponding to the deletion of an account.	`KafkaMessageBuilder`

Connector Operation

Description

Expected Groovy Script Output

CREATE_OBJECT

Produce a Kafka message corresponding to the creation or update of an account. The message key, if set, is used as the account's unique identifier.

If not key is set by the Groovy script, then an attribute mapping must be configured for the__objectId__attribute, and the resulting mapping value is taken as account identifier.

KafkaMessageBuilder

PATCH_OBJECT

Produce a Kafka message corresponding to the update of an account for a custom action configured in ApplicationObjectActivationDefinition.

KafkaMessageBuilder

DELETE_OBJECT

Produce a Kafka message corresponding to the deletion of an account.

KafkaMessageBuilder

Inbound Synchronization Script

The inbound synchronization is configured via the ConsumerDefinition.messageToAccountConverter rule. This rule is executed once for each consumed Kafka message and must return an Account.

Inbound Groovy Context (`KAFKA_MESSAGE`)

The Groovy script receives a single binding:

Context Variable Name	Java Type	Description
`KAFKA_MESSAGE`	`KafkaMessage`	The consumed Kafka message (see below).

KafkaMessage API

The KafkaMessage exposes the following properties:

Property	Java Type	Description
`key`	`String`	The message key, may be null.
`headers`	`Map<String, String>`	The message headers. Header values are decoded as UTF-8 strings. If a header name is repeated, the last occurrence wins.
`header(String name)`	`String`	Convenience method to return the value of a specific header, or `null` if absent.
`value`	`Map<String, Object>`	The deserialized JSON message payload, never null but possibly empty. Only JSON payloads are supported.
`topic`	`String`	The topic the message was consumed from, never null.
`partition`	`Integer`	The partition the message was consumed from, never null.
`offset`	`Long`	The offset of the message within its partition, never null.
`timestamp`	`Instant`	The message timestamp, never null.

Expected Output

The rule must return an Account instance. The returned account must carry the attribute identified by ConsumerDefinition.objectUniqueIdentifier, which ties the message to the account it represents.

The account's object class and kind are imposed by the SynchronizationTaskDefinition configuration, so they do not need to be set by the Groovy script.

Triggering a Deletion via Inbound Synchronization

By convention, to signal that a consumed Kafka message represents the deletion of an account on the remote application, the Groovy rule must set the __syncSituation__ built-in attribute of the returned Account to SyncSituation.DELETED.

When this attribute is set, the synchronization engine detects it and forces a DELETE change type, mirroring the reconciliation "CSV line gone" pattern. The synchronization reaction configured for the DELETED situation is then applied (e.g. UNLINK, DELETE_IDM_OBJECT, etc.)

Examples

Outbound Groovy Script Examples

CREATE_OBJECT Operation

Ruby

// The Person to create. The Groovy binding "ACCOUNT" contains the Person's 
// attributes computed by the attribute mappings
def person = [
        msg_uid: ACCOUNT.uid,
        msg_givenName: ACCOUNT.givenName,
        msg_surname: ACCOUNT.surname,
        msg_birthDate: ACCOUNT.birthDate,
        msg_address: ACCOUNT.address
]

KAFKA.message { topic 'test-provisioning-topic' key ACCOUNT.uid value person }

The above script builds a Kafka message targeting the test-provisioning-topic topic, keyed by the account's uid, and whose JSON payload is the person map. Since a key is set, the key value (ACCOUNT.uid) is used as the account's unique identifier returned to the provisioning service.

DELETE_OBJECT Operation

Ruby

// The Person to delete. The Groovy binding "ACCOUNT_ID" holds the id of the 
// account known by Memority (the usual "ACCOUNT" binding is also available)
// The binding ACTIVATION_SITUATION holds the reason triggering the delete, i.e.
// IM object deleted or inactivated, or application's role lost
def person = [
        msg_uid: ACCOUNT_ID,
        deleted: true,
        reason: ACTIVATION_SITUATION.name()
]

KAFKA.message { topic 'test-provisioning-topic' key ACCOUNT_ID value person }

CREATE_OBJECT with Headers

Ruby

def person = [
        msg_uid: ACCOUNT.uid,
        msg_givenName: ACCOUNT.givenName,
        msg_surname: ACCOUNT.surname
]

KAFKA.message {
    topic 'provisioning-topic'
    key ACCOUNT.uid
    header 'operation', 'CREATE'
    header 'source', 'memority'
    value person
}

Inbound Groovy Script Examples

Basic Message-to-Account Conversion

Ruby

def message = KAFKA_MESSAGE as KafkaMessage
def payload = message.value as Map<String, Object>

Account account = new Account()
account.uid = payload.msg_uid
account.givenName = payload.msg_givenName
account.surname = payload.msg_surname

return account

The objectUniqueIdentifier configured in the ConsumerDefinition must match one of the attributes set on the returned Account (e.g. uid in the above example).

Message-to-Account Conversion with Delete Support

Ruby

// Convert a Kafka message to an Account, supporting a sample "delete" convention
// where if the message's payload contains a "deleted" flag set to true, then the
// Account's __syncSituation__ is set to DELETED to convey this intention.
def message = KAFKA_MESSAGE as KafkaMessage
def payload = message.value as Map<String, Object>

Account account = new Account()
account.uid = payload.msg_uid
account.givenName = payload.msg_givenName
account.surname = payload.msg_surname

if (payload.deleted == true) {
    // Note it is technically not possible for now to use the simple syntax:
    // account.__syncSituation__ = SyncSituation.DELETED
    // So we have to use the syntax wih 'setProperty' below:
    account.setProperty('__syncSituation__', SyncSituation.DELETED)
}

return account

When SyncSituation.DELETED is set, the synchronization engine processes the corresponding account as a deletion, just like a reconciliation "gone CSV line" for a CSV file. The sync reaction associated with the DELETED situation (configured as usual in the Application's SynchronizationReactionDefinition) then determines the action taken on the IM object in reaction to the DELETED situation.

Discriminating on Topic or Headers

Here is an example of a rule consuming from several topics that discriminates on the originating topic or on a header:

Ruby

def message = KAFKA_MESSAGE as KafkaMessage
def payload = message.value as Map<String, Object>

Account account = new Account()
account.uid = payload.uid

if (message.topic == 'identities') {
    account.givenName = payload.firstName
    account.surname = payload.lastName
} else if (message.topic == 'contracts') {
    account.contractId = payload.contractId
    account.startDate = payload.startDate
}

// Discriminate on a header
if (message.header('eventType') == 'DELETE') {
    account.setProperty('__syncSituation__', SyncSituation.DELETED)
}

return account

XML Configuration Example

<?xml version="1.0" encoding="UTF-8"?>
<dmn:KafkaConnectorDefinition xmlns:dmn="http://www.memority.com/domino_sync/1_0"
    xmlns:rule="http://www.memority.com/toolkit/rule/1_0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    id="my-kafka-connector">
   <name>My Kafka Connector</name>
   <description>Kafka Connector for the ACME HR system</description>
   <bootstrapServers>settings:kafka.bootstrap.servers</bootstrapServers>
   <producerDefinition>
     <operationsDefinition>
        <operationDefinitions>
           <operationDefinition>
              <action type="COMPUTE" category="OBJECT" engineType="GROOVY">
               <spec><![CDATA[
def person = [
        uid: ACCOUNT.uid,
        givenName: ACCOUNT.givenName,
        surname: ACCOUNT.surname
]
KAFKA.message { topic 'provisioning-topic' key ACCOUNT.uid value person }
               ]]></spec>
              </action>
              <operation>CREATE_OBJECT</operation>
           </operationDefinition>
           <operationDefinition>
               <action type="COMPUTE" category="OBJECT" engineType="GROOVY">
               <spec><![CDATA[
def person = [
        uid: ACCOUNT_ID,
        deleted: true
]
KAFKA.message { topic 'provisioning-topic' key ACCOUNT_ID value person }
               ]]></spec>
               </action>
               <operation>DELETE_OBJECT</operation>
            </operationDefinition>
         </operationDefinitions>
      </operationsDefinition>
      <defaultTopic>provisioning-topic</defaultTopic>
      <acks>ALL</acks>
      <enableIdempotence>true</enableIdempotence>
      <retries>2</retries>
   </producerDefinition>
   <consumerDefinition>
      <topics>settings:kafka.inbound.topics</topics>
      <messageToAccountConverter type="COMPUTE" category="OBJECT" engineType="GROOVY">
         <spec><![CDATA[
def message = KAFKA_MESSAGE as KafkaMessage
def payload = message.value as Map<String, Object>

Account account = new Account()
account.uid = payload.uid
account.givenName = payload.givenName
account.surname = payload.surname

if (payload.deleted == true) {
    account.setProperty('__syncSituation__', SyncSituation.DELETED)
}

return account
         ]]></spec>
      </messageToAccountConverter>
      <objectUniqueIdentifier>uid</objectUniqueIdentifier>
      <enabled>true</enabled>
      <autoOffsetReset>LATEST</autoOffsetReset>
      <concurrency>1</concurrency>
      <maxPollRecords>500</maxPollRecords>
   </consumerDefinition>
   <transportDefinition>
      <tlsEnabled>true</tlsEnabled>
      <tlsProtocol>TLS_1_3</tlsProtocol>
      <enabledTlsProtocols>
         <enabledTlsProtocol>TLS_1_2</enabledTlsProtocol>
         <enabledTlsProtocol>TLS_1_3</enabledTlsProtocol>
      </enabledTlsProtocols>
      <trustedCertificates>settings:kafka.tls.trusted.certificates</trustedCertificates>
   </transportDefinition>
   <authenticationDefinition xsi:type="dmn:MutualTlsAuthenticationDefinitionType">
      <certificateChain>settings:kafka.mtls.certificate.chain</certificateChain>
      <privateKeyUri>settings:kafka.mtls.private.key</privateKeyUri>
   </authenticationDefinition>
</dmn:KafkaConnectorDefinition>

The corresponding ConnectorDefinition referencing the above KafkaConnectorDefinition:

HTML

<?xml version="1.0" encoding="UTF-8"?>
<dmn:ConnectorDefinition xmlns:dmn="http://www.memority.com/domino_sync/1_0"
    id="my-connid-connector">
   <connectorProperties>
      <connectorProperty>
         <name>kafkaConnectorDefinitionId</name>
         <value>my-kafka-connector</value>
      </connectorProperty>
   </connectorProperties>
   <connectorMetadata>
      <connectorRef>
         <connectorName>com.memority.domino.connector.kafka.KafkaConnector</connectorName>
      </connectorRef>
   </connectorMetadata>
</dmn:ConnectorDefinition>

Topic Resolution Priority (Outbound Provisioning)

The topic a message is ultimately sent to is resolved with the following priority (highest first):

The topic set via KafkaMessageBuilder.topic() in the Groovy script
The OperationDefinition.topic configured for the operation
The ProducerDefinition.defaultTopic

The same priority applies to the partition.

Consumer Group

When listening for Kafka messages in the inbound synchronization case, the consumer group id is automatically derived from the KafkaConnectorDefinition id: memority-<tenant>-<kafka-connector-definition-id>. It is deliberately not configurable so that all consumers of a given Kafka Connector share a single group, ensuring that each message is synchronized exactly once.

Introduction

Limitations

Configuration

General Configuration Principles

KafkaConnectorDefinition Properties

Producer Definition

Operation Definition

Consumer Definition

Transport Definition

Authentication Definition

Mutual TLS Authentication Definition

Writing Groovy Scripts

Outbound Provisioning Scripts

Outbound Groovy Contexts

KafkaMessageBuilder API

Connector Operations

Inbound Synchronization Script

Inbound Groovy Context (KAFKA_MESSAGE)

KafkaMessage API

Expected Output

Triggering a Deletion via Inbound Synchronization

Examples

Outbound Groovy Script Examples

CREATE_OBJECT Operation

DELETE_OBJECT Operation

CREATE_OBJECT with Headers

Inbound Groovy Script Examples

Basic Message-to-Account Conversion

Message-to-Account Conversion with Delete Support

Discriminating on Topic or Headers

XML Configuration Example

Topic Resolution Priority (Outbound Provisioning)

Consumer Group

Inbound Groovy Context (`KAFKA_MESSAGE`)