Mark Payne created NIP-47:
-----------------------------
Summary: 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
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)
