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 |