NIFI-15961 - Add Kafka share-group support to ConsumeKafka

[pvillard31]

Summary

NIFI-15961 - Add Kafka share-group support to ConsumeKafka

Introduce share-group consumption for ConsumeKafka backed by Kafka 4.2's GA share consumer (KIP-932). A new "Group Type" property selects between the existing classic Consumer Group code path (default; preserves all prior behavior) and the new Share Group path, and a new "Acknowledgement Mode" property gates per-record vs. implicit acknowledgement when the share-group path is selected.

• Add a KafkaShareConsumerService API contract and Kafka4ShareConsumerService implementation wrapping org.apache.kafka.clients.consumer.ShareConsumer.
• Add a default KafkaConnectionService.getShareConsumerService() method so third-party connection services remain compatible and opt in to Share Group support explicitly.
• Remove classic-consumer properties rejected by the bundled Kafka client's KafkaShareConsumer before constructing the share consumer, with focused behavioral coverage for the known unsupported properties.
• Update ConsumeKafka.verify() to exercise a Share Group subscription and release sampled records back to the group.
• Keep kafka-clients and the Testcontainers image on 4.3.1; enable Share Group broker configuration on that existing image for integration tests.
• Add unit coverage for polling, explicit and implicit acknowledgement, commit, rollback, close, tombstone handling, processor selection, verification, and acknowledgement-mode wiring.
• Add broker-backed integration coverage proving that an EXPLICIT-mode session commit failure releases and redelivers the same Kafka record; retain IMPLICIT-mode delivery coverage and focused no-per-record-ack unit assertions.

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

• Apache NiFi Jira issue created

Pull Request Tracking

• Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
• Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000
• Pull request contains commits signed with a registered key indicating Verified status

Pull Request Formatting

• Pull Request based on current revision of the main branch
• Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

• Build completed using ./mvnw clean install -P contrib-check
  • JDK 21
  • JDK 25

Focused Verification

• Kafka3ConnectionServiceTest and Kafka4ShareConsumerServiceTest: 24 tests passed
• ConsumeKafkaTest: 20 tests passed
• ConsumeKafkaShareGroupIT: 4 tests passed
• Complete nifi-kafka-3-integration module: 132 tests passed, 6 skipped
• Checkstyle and PMD passed for affected modules using -Pcontrib-check

Licensing

• New dependencies are compatible with the Apache License 2.0 according to the License Policy
• New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

• Documentation formatting appears as expected in rendered files

[pvillard31]

I rebased on latest and I'll add Kafka Share-Group backlog reporting as a follow-up item.

[joewitt]

sorry @pvillard31 looks like it needs another rebase and conflict resolution. I'm happy to dig in and review/test. Seems like a cool new feature

[pvillard31]

thanks @joewitt , rebased

[joewitt]

Thanks

1. Share-group rollback is not proven against a broker

What was true before the PR
Classic ConsumeKafka rollback is a seek to the last committed offset. ConsumeKafkaRebalanceIT and related Docker tests already exercise that the record can be consumed again after a failed session.

What is true with the PR
Share-group rollback is RELEASE (Explicit) or close-and-wait-for-lock (Implicit). Docs and Mockito tests describe that. ConsumeKafkaShareGroupIT only asserts a happy-path consume. Nothing in Docker produces a record, fails the session, and shows the same record coming back.

What I propose specifically and why
Add one IT: Group Type = Share, Acknowledgement Mode = Explicit, produce one record, consume it, fail/roll the session, consume again, assert the same payload (and offset if you want) on SUCCESS. That is the behavior this feature is selling. Implicit redelivery can stay a follow-up because it needs the 5s lock in the test container.

2. The Implicit IT does not show Implicit behavior

What was true before the PR
There was one consume path. Ack was offset commit.

What is true with the PR
testShareGroupConsumesProducedRecord and testShareGroupConsumesProducedRecordWithImplicitAcknowledgement both wait for a FlowFile and assert content. The Implicit test never checks that per-record acknowledge() was skipped, that rollback does not RELEASE, or that a second consume does not immediately see the record.

What I propose specifically and why
Keep the consume assertion, then either (a) Mockito-level: Implicit commit() / rollback() never calls ShareConsumer.acknowledge, or (b) Docker: after SUCCESS, do not produce again, roll or stop, and assert no duplicate until lock expiry. Right now the Implicit test only proves Share Group still consumes.

[joewitt]

Unsupported share-consumer configs (hand-copied deny list)

What was true before the PR

Kafka3ConnectionService built one consumerProperties bag at enable time and used it for classic KafkaConsumer. That bag always includes connectivity (bootstrap, SASL/SSL, timeouts, max.poll.records) and Transaction Isolation Level (isolation.level, default read_committed). Classic-only knobs such as auto.offset.reset and enable.auto.commit are applied later, only on getConsumerService. Users can also add Kafka consumer keys as dynamic properties on the controller service.

Classic KafkaConsumer accepts those keys. Presence is not an error. NiFi already ships one kafka-clients version per release (this tree: 4.3.1). Users still run many broker versions; the default consume path is that classic client. None of this required deleting keys before constructing the consumer.

What is true with the PR

Share Group reuses that same bag to construct KafkaShareConsumer. KIP-932’s share client is a different type. In kafka-clients 4.3.1, ShareConsumerConfig (package-private) walks a private List<String> SHARE_GROUP_UNSUPPORTED_CONFIGS and throws ConfigException if any of those names is present, regardless of value: "<name> cannot be set when using a share group." The ten names today are:

auto.offset.reset, enable.auto.commit, group.instance.id, isolation.level, partition.assignment.strategy, interceptor.classes, session.timeout.ms, heartbeat.interval.ms, group.protocol, group.remote.assignor

That check is new with share groups. It is client-side, before any broker call. It is not a cluster rule that old brokers started enforcing.

Because the shared bag always contains isolation.level, a stock connection service cannot build a share consumer unless that key is removed. The other nine are not applied by getShareConsumerService itself; they appear if someone set them as dynamic properties (or if a later CS property lands in the bag). The PR copies all ten into properties.remove(...) and comments “keep this list in sync with ShareConsumerConfig.SHARE_GROUP_UNSUPPORTED_CONFIGS.” NiFi cannot reference that field: the class is package-private and the list is private. The lists match 4.3.1 today. They will not stay matched by compilation.

This does not couple the default ConsumeKafka path to a Kafka broker version. Group Type still defaults to Consumer Group; mixed 3.x / 4.0 brokers keep working as before. Share Group is opt-in and already documented as 4.1+ / 4.2+ GA brokers. The new coupling is NiFi source duplicating a private deny list inside the bundled kafka-clients jar. A later NiFi kafka-clients bump that adds a rejected key fails at processor start (ConfigException). A bump that later allows a key we still strip means share users cannot set it until NiFi stops removing it. KafkaShareConsumer is also @Evolving, so share behavior can change on a client bump without a broker change.

What I propose specifically and why

Do not try to call Kafka’s constant from production code; it is not part of the public API.

Keep the hardcoded remove()s (they are the only way to reuse the classic bag). Pull them into a package-visible helper, e.g. stripShareUnsupportedConsumerConfigs(Properties), used by getShareConsumerService.

Add a unit test that reads Kafka’s list by reflection and asserts every key is gone after the helper runs:

Class<?> clazz = Class.forName("org.apache.kafka.clients.consumer.ShareConsumerConfig");
Field field = clazz.getDeclaredField("SHARE_GROUP_UNSUPPORTED_CONFIGS");
field.setAccessible(true);
@SuppressWarnings("unchecked")
List<String> kafkaUnsupported = (List<String>) field.get(null);

Properties properties = new Properties();
for (String key : kafkaUnsupported) {
    properties.put(key, "placeholder");
}
stripShareUnsupportedConsumerConfigs(properties);

assertTrue(kafkaUnsupported.stream().noneMatch(properties::containsKey),
    "Share consumer still carries keys Kafka rejects: " + properties.keySet());

That test fails on the next kafka-clients bump that adds a rejected key, instead of failing when someone enables Share Group. Over-stripping (we remove a key Kafka later allows) is the remaining drift; that is acceptable compared with constructor failure, and the same test makes the mismatch visible when the Kafka list changes.

Also drop the “Kafka 4.2+” wording on this block. The reject list is defined by the kafka-clients jar NiFi compiles against (4.3.1 here), not by the broker version in the field.

[joewitt]

What was true before the PR

kafka-clients and the Testcontainers image were already 4.3.1 on main (kafka-clients.version and kafka.docker.image in the parent POM; AbstractKafkaBaseIT / Kafka3ConnectionServiceBaseIT default to apache/kafka:4.3.1). Classic Consume/Publish ITs already ran against that broker. There was nothing to bump to 4.2.0.

What is true with the PR

Share ITs use that same 4.3.1 image. What actually changed is broker config on that image so share groups work: share in group.coordinator.rebalance.protocols, share coordinator enabled, share-coordinator topic RF/ISR for a single-node container, and a short group.share.record.lock.duration.ms (5s) for tests.

The PR body still has a bullet: “Bump the integration-test broker to apache/kafka:4.2.0”. That reads as either a version bump that did not happen or a downgrade from 4.3.1 to 4.2.0. The share product requirement (4.1+ / 4.2+ GA brokers in the field) is a different statement and is already in ConsumeKafka docs; mixing it with “we changed the IT image to 4.2.0” is what is wrong.

What I propose specifically and why

Fix the PR summary bullet (and any leftover “we bumped ITs to 4.2.0” comments) to say: clients and Testcontainers stay on 4.3.1; this change only enables share on that broker. Otherwise reviewers will think the Kafka test matrix moved, and it did not. Broker 4.2 as the minimum for Share Group in production can stay in processor docs; that is KIP-932 GA, not the Docker tag used in CI.

[pvillard31]

Thanks @joewitt for the detailed review. I pushed 25c5f29 with the following changes:

• Added a broker-backed EXPLICIT-mode integration test that simulates a NiFi session commit failure through the actual ConsumeKafka failure callback. It verifies that the released Kafka record is redelivered with the same payload, topic, partition, and offset.
• Renamed the IMPLICIT-mode integration test to clarify that it covers configuration and end-to-end delivery. The existing service unit tests remain the focused proof that implicit commit, rollback, and acknowledge() do not issue per-record Kafka acknowledgements.
• Extracted the unsupported Share Consumer property removal into stripShareUnsupportedConsumerConfigs(Properties).
• Added parameterized coverage for all ten properties rejected by the bundled Kafka 4.3.1 Share Consumer and a test confirming that supported properties are preserved. I intentionally avoided reflection into Kafka's private configuration list, so future Kafka client upgrades still require reviewing this list explicitly.
• Corrected the source comment and PR description: the restriction comes from the bundled kafka-clients implementation, not the broker version. The Kafka clients and Testcontainers image remain on 4.3.1; the integration setup only enables Share Group broker configuration on that image.