NIFI-11259 - Kafka processor refactor

[greyp9]

Background

Previous iterations of support for Kafka client versions in NiFi (1.0, 2.0, 2.6) duplicated code from existing Kafka processors into new Maven modules, adjusted Kafka client library dependencies for the new modules, and adjusted for API differences as needed. The original JIRA associated with NiFi support for Kafka 3 (NIFI-9330), followed this same approach. After discussion, a new approach was chosen, and a new JIRA (NIFI-11259) created.

• Refactor Kafka client library dependencies into a new controller service.
• Expose a service API that was agnostic of any particular Kafka version.
• Create new processor implementations that interacted with Kafka through the service API.

In particular, the new processor module should have no Kafka dependencies. This is expected to ease the transition to future Kafka versions.

• A new 3.0++ controller service might be created to isolate any major API changes to the Kafka client APIs.
• The new PublishKafka and ConsumeKafka processors would need minimal / no changes to enable interactivity with the new controller service.

Other refactoring activities have been undertaken at the same time:

• The new PublishKafka processor is intended as a replacement for the existing PublishKafka_2_6 and PublishKafkaRecord_2_6 processor pair. It provides FlowFile-based or record-based data handling modes, controlled via a per-processor property. Similarly, ConsumeKafka replaces both ConsumeKafka_2_6 and ConsumeKafkaRecord_2_6. This design adjustment reduces the code duplication that was present in the 2.6 processor set.

New Modules

• nifi-kafka-service-api - API contract for KafkaConnectionService, which exposes access to instances of KafkaConsumerService and KafkaProducerService in a manner agnostic to a particular version of Kafka

  • KafkaProducerService - intermediary logical service brokering interactions with the producer APIs of the Kafka client libraries
  • KafkaConsumerService - intermediary logical service brokering interactions with the producer APIs of the Kafka client libraries

• nifi-kafka-service-api-nar - NiFi NAR wrapper for the KafkaConnectionService API contract

• nifi-kafka-3-service - home for Kafka3ConnectionService, which abstracts Kafka dependencies away from the new Kafka processors, and manages runtime connections to a remote Kafka 3 service instance

• nifi-kafka-3-service-nar - NiFi NAR wrapper for Kafka3ConnectionService

• nifi-kafka-processors - home for PublishKafka and ConsumeKafka processors, which allow interactivity with remote Kafka service instances agnostic to a particular Kafka version

• nifi-kafka-nar - NiFi NAR wrapper for the PublishKafka and ConsumeKafka processors

• nifi-kafka-2-6-integration - test bed to establish runtime behavior (testcontainers/kafka) of Kafka 2.6 processors for certain conditions, in order to better replicate those behaviors

• nifi-kafka-3-integration - testing infrastructure to test new processors / controller service while communicating with an actual (testcontainers) Kafka instance

• nifi-kafka-jacoco - module home for configuration to instrument executions of nifi-kafka-bundle unit tests / integration tests, in order to assess test coverage

Notes

• Integration tests are employed as a "first-class citizen" means of testing expected interactions with Kafka instances, running in testcontainers.
  • https://github.com/testcontainers/testcontainers-java
• It is possible to use a single instance of testcontainers/kafka per Maven module, in order to incur the startup/teardown cost only once. The intent is to employ this strategy where feasible.
• Instances of simplelogger.properties have been useful during development, but these would be removed before PR merge.
• I’ve done a significant amount of runtime testing against containerized Kafka instances using the repo/branch below. This resource may be useful for others who want to do runtime testing without the need for fixed Kafka infrastructure.
  • https://github.com/greyp9/kafka-images/tree/NIFI-12194
• There are opportunities to refactor common methods and declarations into the nifi-kafka-shared module; I’ve avoided that during development to ease the process of rebasing to nifi/main.
• It is likely that this set of new components will co-exist with the existing Kafka 2.6 based processors for some period of time.
  • Allow for migration of existing flows to use the new components.
  • Slight behavioral differences might be anticipated during the "burn in" phase of the new components, due to the scope of work.
• Code compatibility with JRE 8 has been targeted, to leave open the option of backporting this work to the support/1.x branch.
• The PR as is should support the following authentication strategies: PLAINTEXT, SSL, SASL_PLAINTEXT, and SASL_SSL. Support for additional authentication strategies could be handled via follow on JIRAs.
• Support for the KafkaRecordSink form factor could be handled via follow on JIRAs.
• Support for Kafka "exactly once" semantics could be handled via follow on JIRAs.
• Migration documentation could be handled via follow on JIRAs.

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 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 mvn clean install -P contrib-check
  • JDK 21

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

[exceptionfactory]

Thanks for the extensive work on putting together these new Kafka components @greyp9!

The number of integration tests provide a solid foundation for verifying the functionality.

I plan to continue going through the changes, but I noted a few general things on an initial review, such as trace logs, property naming, and the Kafka 3 client version.

[joewitt]

Can we just drop the old Kafka processors and go all in on the latest release? At this point the client libraries have stabilized greatly and us supporting the older stuff feels far less important and worthwhile now.

[greyp9]

Can we just drop the old Kafka processors and go all in on the latest release? At this point the client libraries have stabilized greatly and us supporting the older stuff feels far less important and worthwhile now.

It was difficult to tease out all of the business logic from the existing processor set. Some logic is there because of actual user need, and some is there because of the way Kafka and NiFi evolved alongside each other. It seems like the happy path works well for the new components, but not sure about all of the possible runtime permutations.

In my mind, we’re only talking about the 2.6 processor set; the previous implementations aren’t available in the development executable. It does seem like a good idea to stop investing in enhancements to the 2.6 processors. The 2.6 work in this PR is only scaffolding, intended to better understand runtime processor behavior so it can be reproduced in the new components.

The concern for me is that if we drop the 2.6 set as part of this PR, we’ll make things harder for users during the transition period, in case some important edge case is missed. It would also complicate this PR. I’d like to pitch the idea of a transition period where both sets of components exist, to facilitate transition, and to provide a simple fallback if needed. Longer term, it makes a lot of sense for 2.6 to go the way of 2.0.

[joewitt]

@greyp9 Yeah that is fair. Dropping the 2.x stuff is certainly fair in a diff PR.

More relevant for this PR though would be the naming for the Kafka3 components and breaking the cycle of users having to change processors with Kafka changes. We had to do this during the 0x/1x/2x transitions as the client behaviors just weren't stable enough. But I suspect now they are (just my view -others may have different info).

[greyp9]

Agree with your thoughts on the 0x/1x/2x naming.

The expectation is that, with 3+, the processor will not change. It is written to interact with Kafka through a stable controller service API. The controller service implementation will abstract away any changes needed to conform to future versions of Kafka. The controller service in this PR does use the Kafka major version suffix, but that could change.

[joewitt]

I see what you mean. I'm suggesting even needing to swap out the controller service seems problematic.

I like the controller service model and I recall why this direction was taken. But we should be able to call it Kafka and not Kafka3 even there. IF a later client changes behavior we can call that by specific version.

But in any case...I dont feel that strongly. Fine as is just sharing my thoughts.

[greyp9]

I see what you mean. I'm suggesting even needing to swap out the controller service seems problematic.

The jump from 2.6 to 3.0 involved * slight * changes in the NiFi component code (IIRC compatible API parameter types and additional method arguments). Less than an hour to remediate.

It is a good point. Aside from practical considerations (Maven dependencies declared at compile time), it is hard to anticipate the scope of future changes to the client library API. It'll be interesting to get additional perspectives here.

But in any case...I dont feel that strongly. Fine as is just sharing my thoughts.

And thanks for that! :)

[exceptionfactory]

Thanks for the updates @greyp9. I noted a number of minor recommendations, mostly related to property naming.

I realize the property naming has gone in different directions over time. In general, it seems best that when a NiFi property mirrors a Kafka property, the Kafka property name can be used as the NiFi property name, and then use a display name. If the property is strictly for NiFi, like a Record Reader or Writer, then there is no need to have a different property name versus display name. There are a couple places where some names were carried over, so this is a good opportunity to make adjustments. I think the majority of the property names work as defined in the PR, but there are a handful that should be adjusted.

[exceptionfactory]

@greyp9 It looks like the latest changes are failing on integration tests due to property name changes.

[greyp9]

@greyp9 It looks like the latest changes are failing on integration tests due to property name changes.

Thanks; pushed an update.

[exceptionfactory]

Thanks for your patience on this @greyp9. I think it is close to completion for an initial version.

I noted a couple more property and message related recommendations, particularly about the use of reflection in a test class. If you are able to address those comments, This should be ready for merging soon.

[exceptionfactory]

@greyp9 I ran through some integration and runtime tests, evaluating the configuration experience in particular. I adjusted the property ordering in the Processors and Controller Service to follow a logical group of required and then optional, and also keeping groups of dependent properties together.

I made one substantive change to the Transaction Isolation Level property, changing it to a described value instead of a Boolean, which is what I had intended in a previous comment. This way, it reflects the Kafka property value directly, and has corresponding description information.

With these changes, I am planning to merge pending successful automated builds.

[exceptionfactory]

Thanks again for the work on these new components @greyp9! As you noted in the summary, there will be some additional work involved, but this moves things forward and provides an opportunity for additional iterations, as well as subsequent migration away from the 2.6 versions. +1 merging