Repository: nifi-registry Updated Branches: refs/heads/master 7e4e97134 -> cb3b292fa
NIFIREG-194: Updated VersionedConnection to support load balancing configuration NIFIREG-194: Added allowableValues to annotation for the getLoadBalanceCompression() method of VersionedConnection This closes #136. Signed-off-by: Bryan Bende <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi-registry/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi-registry/commit/cb3b292f Tree: http://git-wip-us.apache.org/repos/asf/nifi-registry/tree/cb3b292f Diff: http://git-wip-us.apache.org/repos/asf/nifi-registry/diff/cb3b292f Branch: refs/heads/master Commit: cb3b292fafff500f6e4daa6cdb075d86547131e6 Parents: 7e4e971 Author: Mark Payne <[email protected]> Authored: Tue Aug 14 09:03:38 2018 -0400 Committer: Bryan Bende <[email protected]> Committed: Tue Aug 14 13:47:00 2018 -0400 ---------------------------------------------------------------------- .../nifi/registry/flow/VersionedConnection.java | 36 ++++ .../ConciseEvolvingDifferenceDescriptor.java | 10 +- .../flow/diff/DifferenceDescriptor.java | 4 +- .../nifi/registry/flow/diff/DifferenceType.java | 25 +-- .../flow/diff/EvolvingDifferenceDescriptor.java | 10 +- .../nifi/registry/flow/diff/FlowDifference.java | 4 + .../flow/diff/StandardFlowComparator.java | 170 ++++++++++--------- .../flow/diff/StandardFlowDifference.java | 15 +- .../flow/diff/StaticDifferenceDescriptor.java | 10 +- 9 files changed, 175 insertions(+), 109 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedConnection.java ---------------------------------------------------------------------- diff --git a/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedConnection.java b/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedConnection.java index 4e9fd8e..3a7dd6f 100644 --- a/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedConnection.java +++ b/nifi-registry-data-model/src/main/java/org/apache/nifi/registry/flow/VersionedConnection.java @@ -35,6 +35,11 @@ public class VersionedConnection extends VersionedComponent { private List<String> prioritizers; private List<Position> bends; + private String loadBalanceStrategy; + private String partitioningAttribute; + private String loadBalanceCompression; + + @ApiModelProperty("The source of the connection.") public ConnectableComponent getSource() { return source; @@ -134,6 +139,37 @@ public class VersionedConnection extends VersionedComponent { this.prioritizers = prioritizers; } + @ApiModelProperty(value = "The Strategy to use for load balancing data across the cluster, or null, if no Load Balance Strategy has been specified.", + allowableValues = "DO_NOT_LOAD_BALANCE, PARTITION_BY_ATTRIBUTE, ROUND_ROBIN, SINGLE_NODE") + public String getLoadBalanceStrategy() { + return loadBalanceStrategy; + } + + public void setLoadBalanceStrategy(String loadBalanceStrategy) { + this.loadBalanceStrategy = loadBalanceStrategy; + } + + @ApiModelProperty("The attribute to use for partitioning data as it is load balanced across the cluster. If the Load Balance Strategy is configured to use PARTITION_BY_ATTRIBUTE, the value " + + "returned by this method is the name of the FlowFile Attribute that will be used to determine which node in the cluster should receive a given FlowFile. If the Load Balance Strategy is " + + "unset or is set to any other value, the Partitioning Attribute has no effect.") + public String getPartitioningAttribute() { + return partitioningAttribute; + } + + public void setPartitioningAttribute(final String partitioningAttribute) { + this.partitioningAttribute = partitioningAttribute; + } + + @ApiModelProperty(value = "Whether or not compression should be used when transferring FlowFiles between nodes", + allowableValues = "DO_NOT_COMPRESS, COMPRESS_ATTRIBUTES_ONLY, COMPRESS_ATTRIBUTES_AND_CONTENT") + public String getLoadBalanceCompression() { + return loadBalanceCompression; + } + + public void setLoadBalanceCompression(final String compression) { + this.loadBalanceStrategy = compression; + } + @Override public ComponentType getComponentType() { return ComponentType.CONNECTION; http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConciseEvolvingDifferenceDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConciseEvolvingDifferenceDescriptor.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConciseEvolvingDifferenceDescriptor.java index 9efddf3..76fdfc5 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConciseEvolvingDifferenceDescriptor.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/ConciseEvolvingDifferenceDescriptor.java @@ -30,7 +30,7 @@ public class ConciseEvolvingDifferenceDescriptor implements DifferenceDescriptor @Override public String describeDifference(final DifferenceType type, final String flowAName, final String flowBName, final VersionedComponent componentA, - final VersionedComponent componentB, final Object valueA, final Object valueB) { + final VersionedComponent componentB, final String fieldName, final Object valueA, final Object valueB) { final String description; switch (type) { @@ -41,16 +41,16 @@ public class ConciseEvolvingDifferenceDescriptor implements DifferenceDescriptor description = String.format("%s was removed", componentA.getComponentType().getTypeName()); break; case PROPERTY_ADDED: - description = String.format("Property '%s' was added", valueB); + description = String.format("Property '%s' was added", fieldName); break; case PROPERTY_REMOVED: - description = String.format("Property '%s' was removed", valueA); + description = String.format("Property '%s' was removed", fieldName); break; case VARIABLE_ADDED: - description = String.format("Variable '%s' was added", valueB); + description = String.format("Variable '%s' was added", fieldName); break; case VARIABLE_REMOVED: - description = String.format("Variable '%s' was removed", valueA); + description = String.format("Variable '%s' was removed", fieldName); break; case POSITION_CHANGED: description = "Position was changed"; http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceDescriptor.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceDescriptor.java index 759d9a4..56e65ef 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceDescriptor.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceDescriptor.java @@ -26,9 +26,11 @@ public interface DifferenceDescriptor { * @param type the difference * @param componentA the component in "Flow A" * @param componentB the component in "Flow B" + * @param fieldName the name of the field that changed, or <code>null</code> if the field name does not apply for the difference type * @param valueA the value being compared from "Flow A" * @param valueB the value being compared from "Flow B" * @return a human-readable description of how the flows differ */ - String describeDifference(DifferenceType type, String flowAName, String flowBName, VersionedComponent componentA, VersionedComponent componentB, Object valueA, Object valueB); + String describeDifference(DifferenceType type, String flowAName, String flowBName, VersionedComponent componentA, VersionedComponent componentB, String fieldName, + Object valueA, Object valueB); } http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceType.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceType.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceType.java index 42d5838..047f557 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceType.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/DifferenceType.java @@ -64,16 +64,6 @@ public enum DifferenceType { AUTO_TERMINATED_RELATIONSHIPS_CHANGED("Auto-Terminated Relationships Changed"), /** - * The component has a different set of Outbound Connections in each of the flows - */ - OUTBOUND_CONNECTIONS_CHANGED("Outbound Connections Changed"), - - /** - * The component has a different set of Inbound Connections in each of the flows - */ - INBOUND_CONNECTIONS_CHANGED("Inbound Connections Changed"), - - /** * The component has a different scheduling strategy in each of the flows */ SCHEDULING_STRATEGY_CHANGED("Scheduling Strategy Changed"), @@ -159,6 +149,21 @@ public enum DifferenceType { BACKPRESSURE_DATA_SIZE_THRESHOLD_CHANGED("Backpressure Data Size Threshold Changed"), /** + * The Connection has a different value for the Load Balance Strategy in each of the flows + */ + LOAD_BALANCE_STRATEGY_CHANGED("Load-Balance Strategy Changed"), + + /** + * The Connection has a different value for the Partitioning Attribute in each of the flows + */ + PARTITIONING_ATTRIBUTE_CHANGED("Partitioning Attribute Changed"), + + /** + * The Connection has a different value for the Load Balancing Compression in each of the flows + */ + LOAD_BALANCE_COMPRESSION_CHANGED("Load-Balance Compression Changed"), + + /** * The Connection has a different set of Bend Points in each of the flows */ BENDPOINTS_CHANGED("Connection Bend Points Changed"), http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/EvolvingDifferenceDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/EvolvingDifferenceDescriptor.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/EvolvingDifferenceDescriptor.java index a4d1a65..c3300e6 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/EvolvingDifferenceDescriptor.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/EvolvingDifferenceDescriptor.java @@ -27,7 +27,7 @@ public class EvolvingDifferenceDescriptor implements DifferenceDescriptor { @Override public String describeDifference(final DifferenceType type, final String flowAName, final String flowBName, final VersionedComponent componentA, - final VersionedComponent componentB, final Object valueA, final Object valueB) { + final VersionedComponent componentB, final String fieldName, final Object valueA, final Object valueB) { final String description; switch (type) { @@ -38,16 +38,16 @@ public class EvolvingDifferenceDescriptor implements DifferenceDescriptor { description = String.format("%s with ID %s was removed from flow", componentA.getComponentType().getTypeName(), componentA.getIdentifier()); break; case PROPERTY_ADDED: - description = String.format("Property '%s' was added to %s with ID %s", valueB, componentB.getComponentType().getTypeName(), componentB.getIdentifier()); + description = String.format("Property '%s' was added to %s with ID %s", fieldName, componentB.getComponentType().getTypeName(), componentB.getIdentifier()); break; case PROPERTY_REMOVED: - description = String.format("Property '%s' was removed from %s with ID %s", valueA, componentA.getComponentType().getTypeName(), componentA.getIdentifier()); + description = String.format("Property '%s' was removed from %s with ID %s", fieldName, componentA.getComponentType().getTypeName(), componentA.getIdentifier()); break; case VARIABLE_ADDED: - description = String.format("Variable '%s' was added to Process Group with ID %s", valueB, componentB.getIdentifier()); + description = String.format("Variable '%s' was added to Process Group with ID %s", fieldName, componentB.getIdentifier()); break; case VARIABLE_REMOVED: - description = String.format("Variable '%s' was removed from Process Group with ID %s", valueA, componentA.getIdentifier()); + description = String.format("Variable '%s' was removed from Process Group with ID %s", fieldName, componentA.getIdentifier()); break; default: description = String.format("%s for %s with ID %s from '%s' to '%s'", http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/FlowDifference.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/FlowDifference.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/FlowDifference.java index 50fcbd4..249bf42 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/FlowDifference.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/FlowDifference.java @@ -17,6 +17,8 @@ package org.apache.nifi.registry.flow.diff; +import java.util.Optional; + import org.apache.nifi.registry.flow.VersionedComponent; public interface FlowDifference { @@ -26,6 +28,8 @@ public interface FlowDifference { VersionedComponent getComponentB(); + Optional<String> getFieldName(); + Object getValueA(); Object getValueB(); http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java index de5f097..f98225e 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowComparator.java @@ -39,6 +39,9 @@ import org.apache.nifi.registry.flow.VersionedRemoteGroupPort; import org.apache.nifi.registry.flow.VersionedRemoteProcessGroup; public class StandardFlowComparator implements FlowComparator { + private static final String DEFAULT_LOAD_BALANCE_STRATEGY = "DO_NOT_LOAD_BALANCE"; + private static final String DEFAULT_PARTITIONING_ATTRIBUTE = ""; + private static final String DEFAULT_LOAD_BALANCE_COMPRESSION = "DO_NOT_COMPRESS"; private final ComparableDataFlow flowA; private final ComparableDataFlow flowB; @@ -77,26 +80,21 @@ public class StandardFlowComparator implements FlowComparator { final Set<FlowDifference> differences = new HashSet<>(); - componentMapA.entrySet().stream() - .forEach(entry -> { - final T componentA = entry.getValue(); - final T componentB = componentMapB.get(entry.getKey()); + componentMapA.forEach((key, componentA) -> { + final T componentB = componentMapB.get(key); + comparator.compare(componentA, componentB, differences); + }); + componentMapB.forEach((key, componentB) -> { + final T componentA = componentMapA.get(key); + + // if component A is not null, it has already been compared above. If component A + // is null, then it is missing from Flow A but present in Flow B, so we will just call + // compare(), which will handle this for us. + if (componentA == null) { comparator.compare(componentA, componentB, differences); - }); - - componentMapB.entrySet().stream() - .forEach(entry -> { - final T componentB = entry.getValue(); - final T componentA = componentMapA.get(entry.getKey()); - - // if component A is not null, it has already been compared above. If component A - // is null, then it is missing from Flow A but present in Flow B, so we will just call - // compare(), which will handle this for us. - if (componentA == null) { - comparator.compare(componentA, componentB, differences); - } - }); + } + }); return differences; } @@ -176,73 +174,67 @@ public class StandardFlowComparator implements FlowComparator { final Map<String, VersionedPropertyDescriptor> descriptorsA, final Map<String, VersionedPropertyDescriptor> descriptorsB, final Set<FlowDifference> differences) { - propertiesA.entrySet().stream() - .forEach(entry -> { - final String valueA = entry.getValue(); - final String valueB = propertiesB.get(entry.getKey()); - - VersionedPropertyDescriptor descriptor = descriptorsA.get(entry.getKey()); - if (descriptor == null) { - descriptor = descriptorsB.get(entry.getKey()); + propertiesA.forEach((key, valueA) -> { + final String valueB = propertiesB.get(key); + + VersionedPropertyDescriptor descriptor = descriptorsA.get(key); + if (descriptor == null) { + descriptor = descriptorsB.get(key); + } + + final String displayName; + if (descriptor == null) { + displayName = key; + } else { + displayName = descriptor.getDisplayName() == null ? descriptor.getName() : descriptor.getDisplayName(); + } + + if (valueA == null && valueB != null) { + differences.add(difference(DifferenceType.PROPERTY_ADDED, componentA, componentB, displayName, valueA, valueB)); + } else if (valueA != null && valueB == null) { + differences.add(difference(DifferenceType.PROPERTY_REMOVED, componentA, componentB, displayName, valueA, valueB)); + } else if (valueA != null && !valueA.equals(valueB)) { + // If the property in Flow A references a Controller Service that is not available in the flow + // and the property in Flow B references a Controller Service that is available in its environment + // but not part of the Versioned Flow, then we do not want to consider this to be a Flow Difference. + // This is typically the case when a flow is versioned in one instance, referencing an external Controller Service, + // and then imported into another NiFi instance. When imported, the property does not point to any existing Controller + // Service, and the user must then point the property an existing Controller Service. We don't want to consider the + // flow as having changed, since it is an environment-specific change (similar to how we handle variables). + if (descriptor != null && descriptor.getIdentifiesControllerService()) { + final boolean accessibleA = externallyAccessibleServiceIds.contains(valueA); + final boolean accessibleB = externallyAccessibleServiceIds.contains(valueB); + if (!accessibleA && accessibleB) { + return; + } } + differences.add(difference(DifferenceType.PROPERTY_CHANGED, componentA, componentB, displayName, valueA, valueB)); + } + }); + + propertiesB.forEach((key, valueB) -> { + final String valueA = propertiesA.get(key); + + // If there are any properties for component B that do not exist for Component A, add those as differences as well. + if (valueA == null && valueB != null) { + final VersionedPropertyDescriptor descriptor = descriptorsB.get(key); + final String displayName; if (descriptor == null) { - displayName = entry.getKey(); + displayName = key; } else { displayName = descriptor.getDisplayName() == null ? descriptor.getName() : descriptor.getDisplayName(); } - if (valueA == null && valueB != null) { - differences.add(difference(DifferenceType.PROPERTY_ADDED, componentA, componentB, displayName, displayName)); - } else if (valueA != null && valueB == null) { - differences.add(difference(DifferenceType.PROPERTY_REMOVED, componentA, componentB, displayName, displayName)); - } else if (valueA != null && valueB != null && !valueA.equals(valueB)) { - // If the property in Flow A references a Controller Service that is not available in the flow - // and the property in Flow B references a Controller Service that is available in its environment - // but not part of the Versioned Flow, then we do not want to consider this to be a Flow Difference. - // This is typically the case when a flow is versioned in one instance, referencing an external Controller Service, - // and then imported into another NiFi instance. When imported, the property does not point to any existing Controller - // Service, and the user must then point the property an existing Controller Service. We don't want to consider the - // flow as having changed, since it is an environment-specific change (similar to how we handle variables). - if (descriptor.getIdentifiesControllerService()) { - final boolean accessibleA = externallyAccessibleServiceIds.contains(valueA); - final boolean accessibleB = externallyAccessibleServiceIds.contains(valueB); - if (!accessibleA && accessibleB) { - return; - } - } - - differences.add(difference(DifferenceType.PROPERTY_CHANGED, componentA, componentB, displayName + "=" + valueA, displayName + "=" + valueB)); - } - }); - - propertiesB.entrySet().stream() - .forEach(entry -> { - final String valueA = propertiesA.get(entry.getKey()); - final String valueB = entry.getValue(); - - // If there are any properties for component B that do not exist for Component A, add those as differences as well. - if (valueA == null && valueB != null) { - final VersionedPropertyDescriptor descriptor = descriptorsB.get(entry.getKey()); - - final String displayName; - if (descriptor == null) { - displayName = entry.getKey(); - } else { - displayName = descriptor.getDisplayName() == null ? descriptor.getName() : descriptor.getDisplayName(); - } - - differences.add(difference(DifferenceType.PROPERTY_ADDED, componentA, componentB, displayName, displayName)); - } - }); + differences.add(difference(DifferenceType.PROPERTY_ADDED, componentA, componentB, displayName, null, valueB)); + } + }); } private void compare(final VersionedFunnel funnelA, final VersionedFunnel funnelB, final Set<FlowDifference> differences) { - if (compareComponents(funnelA, funnelB, differences)) { - return; - } + compareComponents(funnelA, funnelB, differences); } private void compare(final VersionedLabel labelA, final VersionedLabel labelB, final Set<FlowDifference> differences) { @@ -257,9 +249,7 @@ public class StandardFlowComparator implements FlowComparator { } private void compare(final VersionedPort portA, final VersionedPort portB, final Set<FlowDifference> differences) { - if (compareComponents(portA, portB, differences)) { - return; - } + compareComponents(portA, portB, differences); } private void compare(final VersionedRemoteProcessGroup rpgA, final VersionedRemoteProcessGroup rpgB, final Set<FlowDifference> differences) { @@ -275,8 +265,8 @@ public class StandardFlowComparator implements FlowComparator { addIfDifferent(differences, DifferenceType.RPG_TRANSPORT_PROTOCOL_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getTransportProtocol); addIfDifferent(differences, DifferenceType.YIELD_DURATION_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getYieldDuration); - differences.addAll(compareComponents(rpgA.getInputPorts(), rpgB.getInputPorts(), (a, b, diffs) -> compare(a, b, diffs))); - differences.addAll(compareComponents(rpgA.getOutputPorts(), rpgB.getOutputPorts(), (a, b, diffs) -> compare(a, b, diffs))); + differences.addAll(compareComponents(rpgA.getInputPorts(), rpgB.getInputPorts(), this::compare)); + differences.addAll(compareComponents(rpgA.getOutputPorts(), rpgB.getOutputPorts(), this::compare)); } private void compare(final VersionedRemoteGroupPort portA, final VersionedRemoteGroupPort portB, final Set<FlowDifference> differences) { @@ -307,7 +297,7 @@ public class StandardFlowComparator implements FlowComparator { addIfDifferent(differences, DifferenceType.VERSIONED_FLOW_COORDINATES_CHANGED, groupA, groupB, VersionedProcessGroup::getVersionedFlowCoordinates); - if (groupA != null && groupB != null && groupA.getVersionedFlowCoordinates() == null && groupB.getVersionedFlowCoordinates() == null) { + if (groupA.getVersionedFlowCoordinates() == null && groupB.getVersionedFlowCoordinates() == null) { differences.addAll(compareComponents(groupA.getConnections(), groupB.getConnections(), this::compare)); differences.addAll(compareComponents(groupA.getProcessors(), groupB.getProcessors(), this::compare)); differences.addAll(compareComponents(groupA.getControllerServices(), groupB.getControllerServices(), this::compare)); @@ -334,6 +324,15 @@ public class StandardFlowComparator implements FlowComparator { addIfDifferent(differences, DifferenceType.PRIORITIZERS_CHANGED, connectionA, connectionB, VersionedConnection::getPrioritizers); addIfDifferent(differences, DifferenceType.SELECTED_RELATIONSHIPS_CHANGED, connectionA, connectionB, VersionedConnection::getSelectedRelationships); addIfDifferent(differences, DifferenceType.SOURCE_CHANGED, connectionA, connectionB, c -> c.getSource().getId()); + + addIfDifferent(differences, DifferenceType.LOAD_BALANCE_STRATEGY_CHANGED, connectionA, connectionB, + conn -> conn.getLoadBalanceStrategy() == null ? DEFAULT_LOAD_BALANCE_STRATEGY : conn.getLoadBalanceStrategy()); + + addIfDifferent(differences, DifferenceType.PARTITIONING_ATTRIBUTE_CHANGED, connectionA, connectionB, + conn -> conn.getPartitioningAttribute() == null ? DEFAULT_PARTITIONING_ATTRIBUTE : conn.getPartitioningAttribute()); + + addIfDifferent(differences, DifferenceType.LOAD_BALANCE_COMPRESSION_CHANGED, connectionA, connectionB, + conn -> conn.getLoadBalanceCompression() == null ? DEFAULT_LOAD_BALANCE_COMPRESSION : conn.getLoadBalanceCompression()); } @@ -366,7 +365,7 @@ public class StandardFlowComparator implements FlowComparator { return; } - differences.add(difference(type, componentA, componentB, valueA, valueB)); + differences.add(difference(type, componentA, componentB, null, valueA, valueB)); } private boolean isEmpty(final Collection<?> collection) { @@ -386,8 +385,15 @@ public class StandardFlowComparator implements FlowComparator { } } - private FlowDifference difference(final DifferenceType type, final VersionedComponent componentA, final VersionedComponent componentB, final Object valueA, final Object valueB) { - final String description = differenceDescriptor.describeDifference(type, flowA.getName(), flowB.getName(), componentA, componentB, valueA, valueB); + private FlowDifference difference(final DifferenceType type, final VersionedComponent componentA, final VersionedComponent componentB, + final Object valueA, final Object valueB) { + return difference(type, componentA, componentB, null, valueA, valueB); + } + + private FlowDifference difference(final DifferenceType type, final VersionedComponent componentA, final VersionedComponent componentB, final String fieldName, + final Object valueA, final Object valueB) { + + final String description = differenceDescriptor.describeDifference(type, flowA.getName(), flowB.getName(), componentA, componentB, fieldName, valueA, valueB); return new StandardFlowDifference(type, componentA, componentB, valueA, valueB, description); } http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowDifference.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowDifference.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowDifference.java index b847a9a..054ee5f 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowDifference.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StandardFlowDifference.java @@ -18,6 +18,7 @@ package org.apache.nifi.registry.flow.diff; import java.util.Objects; +import java.util.Optional; import org.apache.nifi.registry.flow.VersionedComponent; @@ -25,15 +26,22 @@ public class StandardFlowDifference implements FlowDifference { private final DifferenceType type; private final VersionedComponent componentA; private final VersionedComponent componentB; + private final Optional<String> fieldName; private final Object valueA; private final Object valueB; private final String description; public StandardFlowDifference(final DifferenceType type, final VersionedComponent componentA, final VersionedComponent componentB, final Object valueA, final Object valueB, - final String description) { + final String description) { + this(type, componentA, componentB, null, valueA, valueB, description); + } + + public StandardFlowDifference(final DifferenceType type, final VersionedComponent componentA, final VersionedComponent componentB, final String fieldName, + final Object valueA, final Object valueB, final String description) { this.type = type; this.componentA = componentA; this.componentB = componentB; + this.fieldName = Optional.ofNullable(fieldName); this.valueA = valueA; this.valueB = valueB; this.description = description; @@ -55,6 +63,11 @@ public class StandardFlowDifference implements FlowDifference { } @Override + public Optional<String> getFieldName() { + return fieldName; + } + + @Override public Object getValueA() { return valueA; } http://git-wip-us.apache.org/repos/asf/nifi-registry/blob/cb3b292f/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StaticDifferenceDescriptor.java ---------------------------------------------------------------------- diff --git a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StaticDifferenceDescriptor.java b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StaticDifferenceDescriptor.java index 6894923..5bf82c1 100644 --- a/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StaticDifferenceDescriptor.java +++ b/nifi-registry-flow-diff/src/main/java/org/apache/nifi/registry/flow/diff/StaticDifferenceDescriptor.java @@ -30,7 +30,7 @@ public class StaticDifferenceDescriptor implements DifferenceDescriptor { @Override public String describeDifference(final DifferenceType type, final String flowAName, final String flowBName, final VersionedComponent componentA, - final VersionedComponent componentB, final Object valueA, final Object valueB) { + final VersionedComponent componentB, final String fieldName, final Object valueA, final Object valueB) { final String description; switch (type) { @@ -44,19 +44,19 @@ public class StaticDifferenceDescriptor implements DifferenceDescriptor { break; case PROPERTY_ADDED: description = String.format("Property '%s' exists for %s with ID %s in %s but not in %s", - valueB, componentB.getComponentType().getTypeName(), componentB.getIdentifier(), flowBName, flowAName); + fieldName, componentB.getComponentType().getTypeName(), componentB.getIdentifier(), flowBName, flowAName); break; case PROPERTY_REMOVED: description = String.format("Property '%s' exists for %s with ID %s in %s but not in %s", - valueA, componentA.getComponentType().getTypeName(), componentA.getIdentifier(), flowAName, flowBName); + fieldName, componentA.getComponentType().getTypeName(), componentA.getIdentifier(), flowAName, flowBName); break; case VARIABLE_ADDED: description = String.format("Variable '%s' exists for Process Group with ID %s in %s but not in %s", - valueB, componentB.getIdentifier(), flowBName, flowAName); + fieldName, componentB.getIdentifier(), flowBName, flowAName); break; case VARIABLE_REMOVED: description = String.format("Variable '%s' exists for Process Group with ID %s in %s but not in %s", - valueA, componentA.getIdentifier(), flowAName, flowBName); + fieldName, componentA.getIdentifier(), flowAName, flowBName); break; case VERSIONED_FLOW_COORDINATES_CHANGED: if (valueA instanceof VersionedFlowCoordinates && valueB instanceof VersionedFlowCoordinates) {
