environment variable usage in kafka_configs for rdkafka_group input

[shyimo]

Is your feature request related to a problem? Please describe.

It seems like there is no way to pass environment variables for kafka_configs hash in rdkafka_group input.
for example:

<source>
  @type  rdkafka_group
  @label @APPLICATIONS
  topics "#{ENV["APPLICATIONS_TOPIC"] || "application-logs"}"
  format "json"

  kafka_configs {
     "bootstrap.servers": "#{ENV["BROKER_ADDRESS"] || "localhost:9092"}"
  }

</source>

the bootstrap.servers will not parse the BROKER_ADDRESS environment variable.

Describe the solution you'd like

This will be great if there was a way to parse those environment variables inside the kafka_configs hash.
At the current situation, we are not able to fully use all librdkafka features since hard coded values are not possible for some of the fields.

Describe alternatives you've considered

any parsing method that will able to get dynamic fields and not hard coded values

Additional context

No response

[ashie]

Hmm, probably we need to modify fluentd core to realize it.

[shyimo]

@ashie
i ended up doing this:

<source>
  @type  rdkafka_group
  @label @APPLICATIONS
  topics "#{ENV["APPLICATIONS_TOPIC"] || "application-logs"}"
  format "json"

  kafka_configs "#{ENV["KAFKA_CONFIGS"]}"
</source>

and when i run my docker the following:

docker run -e KAFKA_CONFIGS="bootstrap.servers:${KAFKA_BROKERS},group.id:${APPLICATIONS_GROUP_ID}"

the KAFKA_CONFIGS environment variable is parsed as hash object and this works.

[billzhuang]

@shyimo it works to me. Super.

[sendbird-sehwankim]

Here is how I worked it out:

rdkafka_options "#{'{ "security.protocol": "sasl_ssl", "sasl.mechanisms": "SCRAM-SHA-512", "sasl.username": "%s", "sasl.password": "%s" }' % [ENV['MSK_USERNAME'], ENV['MSK_PASSWORD']]}"

[MioOgbeni]

Hi @ashie, what's the status of this issue? I found that you were able to resolve it for the out_rdkafka and out_rdkafka2 module. (https://github.com/fluent/fluent-plugin-kafka/blob/master/ChangeLog#L28) Why does it need modification in FluentD core? (#434 (comment))

Can I try to create a PR that adds this to in_rdkafka_group as well? Or would it be possible to do it generically (at least for all rdkafka modules)? I would need to get this functionality into release as soon as possible.