feat(kafka): support transaction isolation levels

[mbuckley2000]

What

Add an advanced transaction_isolation_level field to both Kafka inputs:

• read_uncommitted (the backward-compatible default)
• read_committed

The legacy kafka input maps it to Sarama's consumer isolation level. kafka_franz maps it to franz-go's fetch isolation level. Both inputs share one field definition and defensive parser, and their generated documentation describes the behavior.

Closes #932.

Why

Both clients support Kafka transactions, but Bento previously offered no way to hide aborted records from consumers of transactional producers such as Kafka Streams exactly-once topologies.

The field name and values follow the existing Redpanda Connect franz-go input convention.

Tests

• Unit/config tests for both inputs:
  • default remains uncommitted;
  • explicit uncommitted;
  • explicit committed;
  • invalid values fail closed.
• Broker-backed integration test for both inputs:
  • open transaction enforces the last-stable-offset barrier;
  • abort releases the following non-transactional record without leaking the aborted one;
  • committed transaction is visible;
  • later aborted transaction is hidden;
  • no extra records leak.
• go vet ./internal/impl/kafka
• golangci-lint -j 1 run -c .golangci.yml ./internal/impl/kafka (0 issues)
• go test ./internal/impl/kafka -run TestIntegrationKafkaTransactionIsolation -count=1 -v -timeout 2m
  • Sarama pass
  • franz-go pass
• Focused generated-document lint passes for both changed pages.

make docs generated the changed Kafka pages successfully; its repository-wide lint subsequently failed on pre-existing DuckDB pages because the normal non-CGO documentation build cannot load the DuckDB driver.

[jem-davies]

Thanks for the PR 🙏 - yeah seems we aren't exposing this config option at the moment.

Only thing I would say is that we could get rid of a lot of stuff:

• The NewStringAnnotatedEnumField will throw lint errors if the value isn't one of the options, anyway. (I do realise it doesn't do this using the .ParseYAML() like in your test) Perhaps this could be simplified such we only have one switch in the components constructor i.e. newFranzKafkaReaderFromConfig - we add a default here that throws an error - like with kafka's sasl.mechanism
• The new unit tests are somewhat redundant - we could keep them - but they largely seem to just be checking configField spec & parsing - not much else.
• I would move this field down such it appears later in the config for doc readability.
• I would just duplicate the field across the two components personally

This review is nit-picky but then we would want to be consistent with the existing patterns in the code imo.

[jem-davies]

Suggested change

	Advanced()
	Advanced().
	Version("1.20.0")