Skip to main content

Celonis Product Documentation

Configuration

The sink offers a set of configuration keys alongside the Kafka Connect defaults for convertors, consumer settings, and so on.

Here is the full list:

Key

Description

Type

Required

Default

connect.ems.endpoint

Contains the EMS API endpoint in the form of: https://[team].[realm].celonis.cloud/continuous-batch-processing/api/v1/[pool-id]/items

STRING

YES

null

connect.ems.authorization.key

Contains the EMS API Authorization header. It should be AppKey <<app-key>> or Bearer <<api-key>>.

STRING

YES

null

connect.ems.target.table

The table in EMS stores the data.

STRING

YES

null

connect.ems.connection.id

Optional parameter. It represents the unique EMS connection identifier.

STRING

NO

null

connect.ems.client.id

An optional parameter representing the client's unique identifier

STRING

NO

null

connect.ems.data.primary.key

The optional parameter contains a list of comma-separated columns, which are the primary keys for the EMS table. If not specified, and the table does not exist, all columns will form the primary key.

STRING

NO

null

connect.ems.order.field.name

Optional parameter used only when primary keys are set. It needs to be a sortable field in the incoming data to allow records deduplication for those sharing the same primary key(s). For details, see the Primary Key(s) section.

STRING

NO

null

connect.ems.data.fallback.varchar.length

An optional parameter representing the STRING (VARCHAR) length when the schema is created in EMS

STRING

NO

null

connect.ems.tmp.dir

The folder stores the temporary files as it accumulates data. If not specified then it uses System.getProperty("java.io.tmpdir").

STRING

NO

System temp directory

connect.ems.commit.size.bytes

The accumulated file maximum size before it is uploaded to EMS. It cannot be less than 1 MB (1000000). A file will be uploaded if the other commit policies are triggered. A file smaller than 1MB can still be uploaded if the records count, the time interval, or a schema change comes first.

LONG

YES

System temp directory

connect.ems.commit.records

The maximum number of records in the accumulated file before it is uploaded to EMS.

INT

YES

null

connect.ems.commit.interval.ms

The time interval in milliseconds to upload the data to EMS if the other two commit policies are not yet applicable. It cannot be less than 1 second.

LONG

YES

null

connect.ems.parquet.write.flush.records

To ensure the file size policy, the number of records after which it should flush the data to the file.

INT

NO

100

connect.ems.error.policy

Specifies the action to be taken if an error occurs while inserting the data. There are three options:* CONTINUE - the error is swallowed* THROW - the error is allowed to propagate. * RETRY - The exception causes the Connect framework to retry the message. The number of retries is set by connecting.ems.max.retries. All errors will be logged automatically, even if the code swallows them.

STRING

NO

THROW

connect.ems.max.retries

The maximum number of times to re-attempt to write the records before the task is marked as failed.

INT

NO

10

connect.ems.obfuscation.fields

An optional value for comma-separated fields to obfuscate. It supports nested values, including arrays.

STRING

NO

null

connect.ems.obfuscation.method

The connector offers 3 types: fix, sha1, and sha512. When a fix is used, the string values are transformed to:***** . For SHA512 a salt is required. See connect.ems.obfuscation.sha512.salt

STRING

NO

fix

connect.ems.obfusation.sha512.salt

Required only when connect.ems.obfuscation.method is set to sha512 and obfuscation fields have been set. If no obfuscation fields have been provided, this configuration is ignored.

STRING

NO

null

connect.ems.debug.keep.parquet.files

For debugging purposes, set the setting to true for the connector to keep the files after they were uploaded.

BOOL

NO

false

connect.ems.proxy.host

The proxy server's hostname, if a proxy is required to access external services.

STRING

NO

null

connect.ems.proxy.port

The port number of the proxy server, if a proxy is required to access external services.

INT

NO

null

connect.ems.proxy.auth.type

If a proxy is required to access external services, the type of proxy to use. There is currently one available option: BASIC - Basic Authentication.

STRING

NO

null

connect.ems.proxy.auth.username

The username for proxy authentication, if a proxy is required to access external services.

STRING

NO

null

connect.ems.proxy.auth.password

The password for proxy authentication, if a proxy is required to access external services.

STRING

NO

null

connect.ems.explode.mode

When each incoming record is a list of records, this will explode (flatten) the records on output. The possible values are: NONELIST - the record must be a List of record types. The sink will discard the List wrapper and write each record. When using this setting with flush counts, the number of exploded records will not be reflected in the flush counts, only the unexploded ones.

STRING

NO

NONE