[NIFI-13408] Dynamically Update Kafka Topic in ConsumeKafka_2_6 Processor

[andrealves23]

Summary

NIFI-13408

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 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 putting together this pull request @andrealves23. For initial review, can you squash the existing commits into a single a commit? That should remove the merge commits and make it easier to start with a clean base set of changes.

[exceptionfactory]

Also of note on the substance, I'm not sure about the implications of this particular change, as the Processor is designed as a "source", not to receive input FlowFiles. There is a large refactor in process for Kafka Processors which should be ready to merge soon, so that is something else to consider.

[exceptionfactory]

Thanks for consolidating the commits @andrealves23.

On closer review, I think it is best to avoid introducing this substantive change to the ConsumeKafka_2_6 Processor.

The general purpose of Consume Processors is to continuously receive data from a given source. Supporting input FlowFiles represents a significant change in behavior, beyond simply making the topic name configurable from an input FlowFile attribute.

Not consuming messages from a particular topic for an extended period of time could result in unexpected processing surges. That is not to say this couldn't work, but it introduces a new layer of complexity. If the project were to consider supporting this type of capability, I think a better approach would be to consider a dedicated Processor. Work on NIFI-11259 introduces a redesigned an modernized approach to Kafka integration, which would provide a much better foundation for anything new. There are still a few things to work out, but it is nearing completion. In light of those forthcoming changes, I recommend revisiting this use case after the new components are merged. That would also provide the opportunity to describe potential use cases in more detail on this Jira issue.

[andrealves23]

Thanks for closer review @exceptionfactory,

The ConsumeKafka_2_6 processor maintains its original behavior even with flowfile support. It can collect Kafka messages regardless of its connection status and without needing stimulation from a flowfile. The key difference is its new capability to support flowfile processing, allowing for dynamic topic changes. I think this change does not increase complexity significantly; it simply adds a new feature while preserving the original behavior.

[joewitt]

@andrealves23 First thank you for contributing. I can appreciate the intent. That said I do agree with the response of @exceptionfactory.

A couple things at play. First is related to the maintainability of the Kafka components. We should be getting rid of the Kafka 2.x components soon in favor of Kafka 3.7/latest or even 4.0 as it will be out very soon. There is a long standing active effort to refactor the Kafka processors underway as well. And of course this is just consume kafka and we also have record oriented processors as well. We can't have behavior deviating too far.

Secondarily there is then the question of this change relative to the purpose of the processor. That processor is written as a source component and designed as such. Switching to allow it to accept flowfiles as input which are effectively signals to change which topic(s) it pays attention to is likely not producing enough of a clear user experience improvement.

As I read this it means the input flowfiles act as signals/switches to change which topic is being monitored. In your intent/flow how will that really work in terms of 'what' creates the input flowfiles and at what rate will new flowfiles arrive? What happens if a new flowfile comes in and there is still more to read on the topic(s) in question/etc.?

This is too complex as-is for a user and changes an otherwise extremely critical/common use component.

To be clear i'm not rejecting the idea. You may well be on to something. I suggest it is offered with a more detailed writeup of the 'why' and how it would work in practice/is intended to work. Then a component that is specific to this DynamicConsumeKafka* pattern. This way we can all offer thoughts on the first part before having to react to the code.

The writeup on the JIRA is a good start but what it lacks is a discussion/description around 'what and how' signaling of topic changes would occur. We need to understand more of that intent/picture.

[exceptionfactory]

Thanks for the additional perspective @joewitt.

@andrealves23 new Kafka Processors and Controller Services are now part of the main branch following the merge of #8463.

As Joe suggested, I recommend writing up a more detailed use case of how this dynamic topic polling would work in the associated Jira issue. Although the code changes may be small, the implications for flow design are very significant. If you can describe the general concept there, and what should drive selecting a topic to poll, that would be very helpful.

With the new KafkaConnectionService API, there is now an abstraction around producing or consuming Kafka records, avoiding tight coupling between NiFi Processors and the Kafka client library version. This new Controller Service API also provides a new way to introduce different Kafka-based Processors.

With that background, I am closing this current pull request, but feel free to continue the discussion on the Jira after evaluating the options based on the new components.

[andrealves23]

I understand your perspectives, @exceptionfactory and @joewitt.

Therefore, as requested, I added more details to the ticket description to explain why it is essential for Apache NiFi to support dynamic Kafka topic management.

I haven't investigated the new Kafka processor code yet, but the idea is to discuss via the Jira ticket whether this feature can genuinely help and improve the user experience. If it is something we decide to move forward with, I will review the new code and determine the best way to add this feature to NiFi.