How can I extract form the logged JSON message a field value that can be used as partition key to the JSON messages to a Kafka topic with filebeat?

[Avi_Chalbani]

How can I extract form the logged JSON message a field value that can be used as partition key to the JSON messages to a Kafka topic with filebeat?

I have an event (JSON message) in file, that needs to be sent to Kafka over filebeat .
A JSON message looks like this:
{"time":1582213700.001,"interval":"2s","worker":11,"application":"1.1.1.1"}

I would like to send this message to Kafka. The partion key should be the application field in the JSON event message. How can I provide the custom application field within the JSON message as partion key for the Kafka record?

The filebeat.yml like this:
…
output.kafka:
version: 0.10.1
hosts: ["{KAFKA1}:9092", "{KAFKA2}:9092", "{KAFKA3}:9092"]
topic: '%{[log_topic]}'
codec.format:
string: '%{[message]}'
key: '%{[message.application]:default}'
partition.hash:
hash:
random: true # if false non-hashable events will be dropped
required_acks: 1
compression: none

https://www.elastic.co/guide/en/beats/libbeat/6.8/config-file-format-type.html#_format_string_sprintf
According to this reference we can refer to event field values by using the format string spec.

With this configuration the default message 'default' is always reported as key. How can I configure the filebeat.yml to extract the custom application field and use this information as Kafka partion key?

Furthermore I have tried to define a filed in the input section as follows:
type: log
enabled: true
paths:
- /var/logs/*event.log
fields:
log_topic: "event"
application: '%{[application]} string'
fields_under_root: true

and a corresponding kafka output as:
output.kafka:
version: 0.10.1
hosts: ["{KAFKA1}:9092", "{KAFKA2}:9092", "{KAFKA3}:9092"]
topic: '%{[log_topic]}'
codec.format:
string: '%{[message]}'
key: '%{[application]:default}'

  partition.hash:
    hash: []
    random: true # if false non-hashable events will be dropped
  required_acks: 1

But then the kafka partion key was always: %{[application]} string