[NopeCharon]

What this PR does / why we need it:

Adds a kafka input to ClusterInput, for Fluent Bit's kafka input plugin. It subscribes to Kafka topics and turns each message into a record. Until now it was only usable through customPlugin.

spec:
  kafka:
    tag: kafka.logs
    brokers: kafka:9092
    topics: app-logs
    format: json
    rdkafka:
      auto.offset.reset: earliest

This renders as:

[Input]
    Name    kafka
    tag    kafka.logs
    brokers    kafka:9092
    topics    app-logs
    format    json
    rdkafka.auto.offset.reset    earliest

The input covers brokers, topics, groupId, clientId, format, pollMs, pollTimeoutMs, bufferMaxSize, enableAutoCommit and threaded, plus any rdkafka.* consumer property. It works like the Kafka output: rdkafkaSecrets reads properties such as sasl.username / sasl.password from a Secret, and if a property is set in both places, the Secret value wins.

Fluent Bit won't start the input without brokers (config: no brokers defined) or topics (config: no topics specified), and it only accepts none or json as format. The CRD therefore requires brokers and topics (MinLength=1) and limits format to those two values. The API server rejects these mistakes before Fluent Bit ever sees the config.

The deepcopy code, CRDs, setup.yaml and API docs are regenerated, and the input is listed in docs/plugins/fluentbit/index.md.

Which issue(s) this PR fixes:

Fixes #2142

Does this PR introduced a user-facing change?

Add the kafka input plugin to the Fluent Bit ClusterInput.

Additional documentation, usage docs, etc.:

New docs/plugins/fluentbit/input/kafka.md

Testing

• TestInput_Kafka_Params and TestInput_Kafka_ParamsOnlySetFields check the rendered parameters with every field set and with only some set, including explicit 0 / false values.
• TestInput_Kafka_ParamsRdkafkaSecrets checks that properties are read from a Secret, that a Secret value overrides a plain value, and that a missing key returns an error.
• TestClusterInputList_Load_Kafka checks the classic and YAML output of a ClusterInput.
• make generate manifests docs-update leaves no diff. make lint reports 0 issues, and go test ./apis/... ./pkg/... passes.
• I tested both rendered configs (classic and YAML, every field set, threaded: true) on fluent/fluent-bit:5.2.0 against a single-node apache/kafka:3.9.1 broker. The input consumed both topics. JSON payloads were parsed into maps, and a non-JSON payload was kept as a string, e.g. {"topic":"app-logs","partition":0,"offset":0,"error":null,"key":null,"payload":{"msg":"hello","level":"info"}}.

This PR and #2145 (prometheus_remote_write input) each add a field next to execWasi in InputSpec. Whichever merges second will need its generated files regenerated.