[jira] [Assigned] (NIP-47) Allow Connectors to more effectively evolve their flows on startup

["Mark Payne (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/NIP-47?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Mark Payne reassigned NIP-47:
-----------------------------

    Assignee: Mark Payne

> Allow Connectors to more effectively evolve their flows on startup
> ------------------------------------------------------------------
>
>                 Key: NIP-47
>                 URL: https://issues.apache.org/jira/browse/NIP-47
>             Project: NiFi Improvement Proposal
>          Issue Type: Improvement
>            Reporter: Mark Payne
>            Assignee: Mark Payne
>            Priority: Major
>
> h1. Motivation
> Connectors provide a way to programmatically control a full NiFi flow. This is a very powerful concept. A missing component here, though, is the ability to effectively manage the case where a startup of NiFi results in a new version of the Connector, and the Connector has a new version of the flow where Connections were removed.
> In such a case, the Connector is forced to maintain the old flow in some way and cause the data to drain using the old version of the flow, and only then migrate to the new version. This is awkward and requires significant effort and is error-prone.
> h1. Proposal
> My proposal, then, is to offer the ability for a Connector to relocate the FlowFiles that exist in a given (no-longer-existing) Connection to a new Connection, possibly transforming the Attributes along the way.
> h2. API
> My intention would be to introduce a new method to the {{Connector}} interface:
> {code:java}
> /**
>  * Provides the opportunity for a Connector to evolve its flow from an earlier version to the current version
>  * by informing the framework how existing data should be handled.
>  *
>  * @param context the context that provides the ability to evolve the flow from its pre-existing version to its current version
>  */
> default void evolveFlow(FlowEvolutionContext context) {
> }{code}
> Where {{FlowEvolutionContext}} is defined as:
> {code:java}
> /**
>  * The FlowEvolutionContext provides the mechanism by which Connectors are able to evolve their flows from
>  * prior versions of the Connectors' flows to the current version.
>  */
> public interface FlowEvolutionContext {
>     /**
>      * Instructs the framework to rehome any FlowFiles that are currently homed to the given removed Connection ID
>      * to the Connection with the given destination Connection ID. Note that invoking this method does not itself necessarily
>      * transfer any FlowFiles between connections. Rather, it informs the framework that the move needs to happen.
>      * It is up to the framework to make the determination when this should happen. The framework may choose to move
>      * the FlowFiles inline during this method invocation, or it may choose to do so later. The framework does guarantee,
>      * however, that this move happens before the Connector is able to start.
>      *
>      * @param removedConnectionId the ID of the Connection in the previous flow
>      * @param destinationConnectionId the ID of the Connection in the current flow
>      * @throws IllegalStateException if no Connection exists in the current flow with the given ID,
>      *   or if a Connection still exists in the current flow with an ID matching the removedConnectionId.
>      */
>     default void relocateFlowFiles(String removedConnectionId, String destinationConnectionId) {
>         relocateFlowFiles(removedConnectionId, destinationConnectionId, AttributesTransformer.NOP);
>     }
>     /**
>      * Instructs the framework to rehome any FlowFiles that are currently homed to the given removed Connection ID
>      * to the Connection with the given destination Connection ID, transforming the FlowFile's attributes as defined by the given
>      * transformer. Note that invoking this method does not itself necessarily
>      * transfer any FlowFiles between connections. Rather, it informs the framework that the move needs to happen.
>      * It is up to the framework to make the determination when this should happen. The framework may choose to move
>      * the FlowFiles inline during this method invocation, or it may choose to do so later. The framework does guarantee,
>      * however, that this move happens before the Connector is able to start.
>      *
>      * @param removedConnectionId the ID of the Connection in the previous flow
>      * @param destinationConnectionId the ID of the Connection in the current flow
>      * @param attributesTransformer a transformer that can be used to manipulate the attributes of FlowFiles that are moved
>      * @throws IllegalStateException if no Connection exists in the current flow with the given ID,
>      *   or if a Connection still exists in the current flow with an ID matching the removedConnectionId.
>      */
>     default void relocateFlowFiles(String removedConnectionId, String destinationConnectionId, AttributesTransformer attributesTransformer) {
>         relocateFlowFiles(removedConnectionId, destinationConnectionId, flowFile -> true, attributesTransformer);
>     }
>     /**
>      * Instructs the framework to rehome any FlowFiles that are currently homed to the given removed Connection ID, and that match the given Predicate,
>      * to the Connection with the given destination Connection ID. The FlowFiles' attributes will be transformed according to the provided AttributesTransformer.
>      * Note that invoking this method does not itself necessarily transfer any FlowFiles between connections. Rather, it informs the framework
>      * that the move needs to happen. It is up to the framework to make the determination when this should happen. The framework may choose to move
>      * the FlowFiles inline during this method invocation, or it may choose to do so later. The framework does guarantee, however, that this move
>      * happens before the Connector is able to start.
>      *
>      * <p>Any FlowFile that does not match the given predicate is ignored. Many invocations may be used along with different Predicates in order to
>      * route FlowFiles from the given Connection to one of multiple destination connections.
>      *
>      * @param removedConnectionId the ID of the Connection in the previous flow
>      * @param destinationConnectionId the ID of the Connection in the current flow
>      * @param filter a predicate that determines whether or not a given FlowFile will be rehomed.
>      * @param attributesTransformer a transformer that can be used to manipulate the attributes of FlowFiles that are moved
>      * @throws IllegalStateException if no Connection exists in the current flow with the given ID,
>      *   or if a Connection still exists in the current flow with an ID matching the removedConnectionId.
>      */
>     void relocateFlowFiles(String removedConnectionId, String destinationConnectionId, Predicate<FlowFile> filter, AttributesTransformer attributesTransformer);
>     /**
>      * Instructs the framework to permanently drop any FlowFiles that are currently homed to the given removed Connection ID
>      *
>      * @param removedConnectionId the ID of the Connection in the previous flow
>      * @throws IllegalStateException if a Connection still exists in the current flow with an ID matching the removedConnectionId.
>      */
>     default void dropFlowFiles(String removedConnectionId) {
>         dropFlowFiles(removedConnectionId, flowFile -> true);
>     }
>     /**
>      * Instructs the framework to permanently drop any FlowFiles that are currently homed to the given removed Connection ID if and only if
>      * they match the given predicate
>      *
>      * @param removedConnectionId the ID of the Connection in the previous flow
>      * @param filter a predicate that determines whether or not a given FlowFile will be dropped.
>      * @throws IllegalStateException if a Connection still exists in the current flow with an ID matching the removedConnectionId.
>      */
>     void dropFlowFiles(String removedConnectionId, Predicate<FlowFile> filter);
> }
> {code}
> And {{AttributesTransformer}} is a simple functional interface:
> {code:java}
> @FunctionalInterface
> public interface AttributesTransformer {
>     Map<String, String> transformAttributes(FlowFile flowFile);
>     AttributesTransformer NOP = FlowFile::getAttributes;
> } {code}
> The method on {{Connector}} would be defaulted to do nothing, so it is backward compatible. But it would allow for scenarios such as "I used to have Processor A -> B -> C -> D, but now I just want to do A -> X -> D" by allowing the data from connection B-C and the data from C-D to be moved to connection A-X:
> {code:java}
> public void evolveFlow(FlowEvolutionContext context) {
>   context.relocateFlowFiles( "B-C", "A-X" );
>   context.relocateFlowFiles( "C-D", "A-X", flowFile -> {
>     final Map<String, String> evolvedAttributes = new HashMap<>(flowFile.getAttributes();
>     evolvedAttributes.remove("mime.type");
>     evolvedAttributes.put("attributeA", "valueA");
>     return evolvedAttributes;
>   });
> } {code}
> The {{Predicate}} allows us to also relocate different FlowFiles to different Connections based on their attributes, size, etc.
> This approach means that we don't have to maintain old versions of the flow or deal with starting up with that old version, draining, and then updating the flow. Instead, we can simply have it move the data to a new Connection.
> As always, the intent of the NIP is to proposal the concept; the API may end up deviating from this slightly, and will go through the normal review-then-commit policy.
> h2. Provenance
> One important gotcha here is that the Provenance lineage could be very confusing, even though it is technically correct in terms of what happened. To account for this, we should introduce a new Provenance Event Type, such as {{RELOCATED}} . This would make it much clearer what happened.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)