You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2020/02/10 19:16:03 UTC
[GitHub] [nifi] markap14 opened a new pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
markap14 opened a new pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045
…Queue Partitions in the constructor, it added the local partition as the first element in that list. This list should be ordered the same across all nodes in the cluster. By making the local partition the first in the array, each node had a different ordering of these partitions. As a result, Partition by Attribute strategy would constantly rebalance flowfiles that it received to other node, and Single Node always transferred data to the first partition, which was the local node, instead of whichever node should have been the first in the list. This commit addresses this issue by instead inserting the local partition intot he 'queuePartitions' array based on the local node identifier.
Thank you for submitting a contribution to Apache NiFi.
Please provide a short description of the PR here:
#### Description of PR
_Enables X functionality; fixes bug NIFI-YYYY._
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [ ] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [ ] Does your PR title start with **NIFI-XXXX** where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [ ] Has your PR been rebased against the latest commit within the target branch (typically `master`)?
- [ ] Is your initial contribution a single, squashed commit? _Additional commits in response to PR reviewer feedback should be made on this branch and pushed to allow change tracking. Do not `squash` or use `--force` when pushing to allow for clean monitoring of changes._
### For code changes:
- [ ] Have you ensured that the full suite of tests is executed via `mvn -Pcontrib-check clean install` at the root `nifi` folder?
- [ ] Have you written or updated unit tests to verify your changes?
- [ ] Have you verified that the full build is successful on both JDK 8 and JDK 11?
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the `LICENSE` file, including the main `LICENSE` file under `nifi-assembly`?
- [ ] If applicable, have you updated the `NOTICE` file, including the main `NOTICE` file found under `nifi-assembly`?
- [ ] If adding new Properties, have you added `.displayName` in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377812741
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
+ if (!partitionList.contains(localPartition)) {
Review comment:
Perfect, I understand. Thank you very much for your explanation. I am very interested in this commit to be able to take the version to production.**+1**
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] asfgit closed pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] markap14 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377806864
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
Review comment:
Yes, `clusterCoordinator.getLocalNodeIdentifier()` can return `null`. In that case, the local partition will never be added in the loop.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] pvillard31 commented on issue #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
pvillard31 commented on issue #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#issuecomment-584797494
Merged into master, thanks @markap14
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] eduardofontes commented on issue #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
eduardofontes commented on issue #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#issuecomment-584783372
PR built and tested with Single Node in a 3 node cluster. +1
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377791523
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
+ if (!partitionList.contains(localPartition)) {
Review comment:
on the other hand, the order would not be altered if it happens that the partition is not in the list for one of the nodes?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377812936
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
Review comment:
all clear.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] markap14 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377833917
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
+ if (!partitionList.contains(localPartition)) {
Review comment:
No problem, glad I was able to clarify.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377380652
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
Review comment:
Mark, to understand why you make sure the local partition is on the list. it is not assumed that in the iteration on the array sortedNodeIdentifiers the local partition should always be added when the id node is localNodeId?.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377732154
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
Review comment:
on the other hand, the order would not be altered if it happens that the partition is not in the list for one of the nodes?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377383687
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
Review comment:
¿Is there a case in which this conditional is not met?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
ivan1221 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377732154
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
Review comment:
on the other hand, the order would not be altered if it happens that the partition is not in the list for one of the nodes?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [nifi] markap14 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
Posted by GitBox <gi...@apache.org>.
markap14 commented on a change in pull request #4045: NIFI-7117: When SocketLoadBalancedFlowFileQueue creates its array of …
URL: https://github.com/apache/nifi/pull/4045#discussion_r377809799
##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
##########
@@ -151,15 +151,19 @@ public SocketLoadBalancedFlowFileQueue(final String identifier, final Connection
// that is not the local node identifier. If the Local Node Identifier is not yet known, that's okay. When it becomes known,
// the queuePartitions array will be recreated with the appropriate partitions.
final List<QueuePartition> partitionList = new ArrayList<>();
- partitionList.add(localPartition);
final NodeIdentifier localNodeId = clusterCoordinator.getLocalNodeIdentifier();
for (final NodeIdentifier nodeId : sortedNodeIdentifiers) {
if (nodeId.equals(localNodeId)) {
- continue;
+ partitionList.add(localPartition);
+ } else {
+ partitionList.add(createRemotePartition(nodeId));
}
+ }
- partitionList.add(createRemotePartition(nodeId));
+ // Ensure that our list of queue partitions always contains the local partition.
+ if (!partitionList.contains(localPartition)) {
Review comment:
You're correct. If local node id ends up being null, we will add localPartition to the end of this, which means that this node will have the order wrong. As the code comment above explains, though, that's okay. The order will only be wrong until the local node identifier is determined. This will generally take milliseconds. Perhaps a few seconds. Then the ordering will be fixed in the `onLocalNodeIdentifierSet` method. In the meantime, it may end up sending the data to the wrong node, and the receiving node would then take that data and re-send it to who it believes to be the correct node in the cluster. It is inefficient, but should only occur for a very short period of time and only when the local node id is null (and this happens rarely, as it's basically a race condition that can happen when the node is added to the cluster for the first time. The next time the cluster is restarted, the local node id should be known from local state.)
The problem previously was that the localPartition was *always* added as the first entry, even when the local node identifier was already known. As a result, it didn't get re-sorted. Now, it will be added only when the local nod id is null and therefore it will be re-sorted when the local node id is known.
Sorry for the huge explanation. I hope it all makes sense.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services