Allow passing advanced librdkafka options to the kafka source and sink

[ghost]

librdkafka which underlies kafka source and sink supports many advanced configuration options.

Some users might have a reason to tune some of them. I propose to add a new advanced configuration option called librdkafka_options to the configuration of both of them.

Its usage could look like this:

[sinks.my_sink_id]
  # REQUIRED - General
  type = "kafka" # must be: "kafka"
  inputs = ["my-source-id"] # example
  bootstrap_servers = "10.14.22.123:9092,10.14.23.332:9092" # example
  key_field = "user_id" # example
  topic = "topic-1234" # example
  # REQUIRED - requests
  encoding = "json" # example, enum
  # OPTIONAL - General
  healthcheck = true # default
  [sinks.my_sink_id.librdkafka_options]
    "max.in.flight.requests.per.connection" = "1000000"
    "max.poll.interval.ms" = "300000"
    "socket.timeout.ms" = "60000"

For example, ClickHouse, which uses librdkafka as the base of its Kafka engine too, allows to pass arbitrary options to it.

Related to #1818.

[binarylogic]

I don't dislike this, but I think it's worth discussing mapping these options ourselves. For obvious reasons:

1. By exposing this to our users we will be even more locked into librdkafka. I'm not that worried about this, because it's unlikely we'll move away.
2. I'd still like to document the available options so they are searchable, etc. And if we're going to do that, it probably makes sense to explicitly support them in the code too.

The other question: how often are these options changing?

[ghost]

I don't dislike this, but I think it's worth discussing mapping these options ourselves

I'm fine with mapping selected options ourselves instead. It has the benefit of supporting type checks for the values, so integer options would be required to be integers and enum options would be required to be enums.

The main question is what options should we map out of more than a hundred of supported options, but we can map only the options which we see demand for. For example, we can start from exposing socket.timeout.ms and message.timeout.ms which are related to #1818.

Actually, we already have some advanced options mapped, such as commit.interval.ms and session.timeout.ms for kafka source.

The other question: how often are these options changing?

From the history looks like mostly just new options are added, but the old ones are not removed.

[binarylogic]

The main question is what options should we map out of more than a hundred of supported options

I'm leaning towards a hybrid approach. Let's map the common options that help to deliver a good UX for the common case, then we could provide a librdkafka_options table as an escape hatch. This at least communicates to the user that these are "more advanced" options.

[binarylogic]

And I don't know enough about Kafka to say which ones are common. The 2 you listed should probably be mapped, and I'd use your best judgment for any others.