At Kafka WIH executeWorkItem, there are 2 nested try-catch blocks, that are not strictly needed, and currently if there is an exception in the producer (inner catch) there is no invocation to handleException method.

{{producer.send }}call is asynchronous and returns a Future<MetadataRecord>: therefore, to complete the workitem, it's needed to wait for the result.

Besides, results map in the failure case is never sent (no completeWorkitem call) and for the sunny scenario, it doesn't add any meaningful information.