In the deployment descriptor runtimeManager should be used to use async command by kafka:

new KafkaWorkItemHandler.KafkaWorkItemHandler(env['kafka_server'], "jbpm_id", "org.apache.kafka.common.serialization.StringSerializer", "org.apache.kafka.common.serialization.StringSerializer", classLoader, runtimeManager);

also added global env set for next properties for all producers created.

org.jbpm.process.workitem.kafka.reconnect.backoff.max.msreconnect.backoff.max.ms "The maximum amount of time in milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect. If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum. After calculating the backoff increase, 20% random jitter is added to avoid connection storms."

org.jbpm.process.workitem.kafka.reconnect.backoff.ms "The base amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly connecting to a host in a tight loop. This backoff applies to all connection attempts by the client to a broker."

org.jbpm.process.workitem.kafka.request.timeout.ms "The configuration controls the maximum amount of time the client will wait for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted."

org.jbpm.process.workitem.kafka.retries "Setting a value greater than zero will cause the client to resend any request that fails with a potentially transient error. It is recommended to set the value to either zero or `MAX_VALUE` and use corresponding timeout parameters to control how long a client should retry a request."

org.jbpm.process.workitem.kafka.retry.backoff.ms "The amount of time to wait before attempting to retry a failed request to a given topic partition. This avoids repeatedly sending requests in a tight loop under some failure scenarios."

org.jbpm.process.workitem.kafka.enable.idempotence "When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream. If 'false', producer retries due to broker failures, etc., may write duplicates of the retried message in the stream. Note that enabling idempotence requires <code> MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION </code> to be less than or equal to 5, RETRIES_CONFIG to be greater than 0 and <code>ACKS_CONFIG </code> must be 'all'. If these values are not explicitly set by the user, suitable values will be chosen. If incompatible values are set, a <code>ConfigException</code> will be thrown."