[nifi] branch main updated: NIFI-12122 Corrected test for Parameter Context updating
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 61c856a54c NIFI-12122 Corrected test for Parameter Context updating
61c856a54c is described below
commit 61c856a54c56a1180a621bc8f17f0ce63044fcc8
Author: exceptionfactory
AuthorDate: Fri Oct 13 22:55:53 2023 -0500
NIFI-12122 Corrected test for Parameter Context updating
Signed-off-by: David Handermann
---
.../flow/synchronization/StandardVersionedComponentSynchronizerTest.java | 1 +
1 file changed, 1 insertion(+)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java
index 0f508004a3..b663517bbf 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizerTest.java
@@ -1131,6 +1131,7 @@ public class StandardVersionedComponentSynchronizerTest {
final VersionedParameterContext parameterContext = new
VersionedParameterContext();
parameterContext.setName("My Params");
+parameterContext.setDescription("Context Description");
parameterContext.setParameters(Collections.emptySet());
final Map parameterContextMap = new
HashMap<>();
[nifi] branch support/nifi-1.x updated: NIFI-12122 Corrected TestFlowController references
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new 1cb7614b0b NIFI-12122 Corrected TestFlowController references
1cb7614b0b is described below
commit 1cb7614b0b5fe02ed883e3fcfaf7212aa5f20984
Author: Joe Gresock
AuthorDate: Fri Oct 6 13:33:23 2023 -0500
NIFI-12122 Corrected TestFlowController references
- Adjusted XmlFlowSynchronizer method invocation for backported changes
Signed-off-by: David Handermann
---
.../src/main/java/org/apache/nifi/controller/XmlFlowSynchronizer.java | 2 +-
.../src/test/java/org/apache/nifi/controller/TestFlowController.java| 2 ++
2 files changed, 3 insertions(+), 1 deletion(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/XmlFlowSynchronizer.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/XmlFlowSynchronizer.java
index ddb86cfafb..1248688a25 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/XmlFlowSynchronizer.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/XmlFlowSynchronizer.java
@@ -574,7 +574,7 @@ public class XmlFlowSynchronizer implements
FlowSynchronizer {
parameterProviderConfiguration = new
StandardParameterProviderConfiguration(configurationDTO.getParameterProviderId(),
configurationDTO.getParameterGroupName(),
configurationDTO.getSynchronized());
}
-final ParameterContext context =
flowManager.createParameterContext(dto.getId(), dto.getName(), parameters,
referencedIds, parameterProviderConfiguration);
+final ParameterContext context =
flowManager.createParameterContext(dto.getId(), dto.getName(),
dto.getDescription(), parameters, referencedIds,
parameterProviderConfiguration);
context.setDescription(dto.getDescription());
return context;
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java
index fc0e4b7bd1..db281d5b7f 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestFlowController.java
@@ -513,6 +513,8 @@ public class TestFlowController {
final String flow = IOUtils.toString(new FileInputStream(flowFile),
StandardCharsets.UTF_8);
final DataFlow proposedDataFlow = new
StandardDataFlow(flow.getBytes(StandardCharsets.UTF_8), null,
authFingerprint.getBytes(StandardCharsets.UTF_8), Collections.emptySet());
+final VersionedFlowSynchronizer flowSynchronizer = new
VersionedFlowSynchronizer(extensionManager,
+nifiProperties.getFlowConfigurationFile(), new
FlowConfigurationArchiveManager(nifiProperties));
try {
controller.synchronize(flowSynchronizer, proposedDataFlow,
mock(FlowService.class), BundleUpdateStrategy.IGNORE_BUNDLE);
controller.initializeFlow();
[nifi] 02/02: NIFI-12160 Kafka Connect: Check for NAR unpacking before starting
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 19153a7609f7c1a4a229d3cf9186903f1b2da86a
Author: Peter Gyori
AuthorDate: Tue Oct 3 14:42:29 2023 +0200
NIFI-12160 Kafka Connect: Check for NAR unpacking before starting
Check that required NAR files are unpacked completely before starting the
Kafka Connector
This closes #7832
Signed-off-by: David Handermann
(cherry picked from commit b2e3898e171b89e25b5e06f6ee849bb3edabbf9a)
---
.../kafka/connect/WorkingDirectoryUtilsTest.java | 222 +
.../kafka/connect/StatelessKafkaConnectorUtil.java | 10 +-
.../nifi/kafka/connect/WorkingDirectoryUtils.java | 100 ++
3 files changed, 329 insertions(+), 3 deletions(-)
diff --git
a/nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java
b/nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java
new file mode 100644
index 00..e98fcaabb4
--- /dev/null
+++
b/nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.kafka.connect;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.io.CleanupMode.ALWAYS;
+
+public class WorkingDirectoryUtilsTest {
+
+@Test
+public void testDeleteNonexistentFile(@TempDir(cleanup = ALWAYS) File
tempDir) {
+File nonexistentFile = new File(tempDir, "testFile");
+
+WorkingDirectoryUtils.purgeDirectory(nonexistentFile);
+
+assertFalse(nonexistentFile.exists());
+}
+
+@Test
+public void testDeleteFlatFile(@TempDir(cleanup = ALWAYS) File tempDir)
throws IOException {
+File file = new File(tempDir, "testFile");
+file.createNewFile();
+
+WorkingDirectoryUtils.purgeDirectory(file);
+
+assertFalse(file.exists());
+}
+
+@Test
+public void testDeleteDirectoryWithContents(@TempDir(cleanup = ALWAYS)
File tempDir) throws IOException {
+File directory = new File(tempDir, "directory");
+File subDirectory = new File(directory, "subDirectory");
+File subDirectoryContent = new File(subDirectory,
"subDirectoryContent");
+File directoryContent = new File(directory, "directoryContent");
+
+directory.mkdir();
+subDirectory.mkdir();
+subDirectoryContent.createNewFile();
+directoryContent.createNewFile();
+
+WorkingDirectoryUtils.purgeDirectory(directory);
+
+assertFalse(directory.exists());
+}
+
+@Test
+public void testPurgeUnpackedNarsEmptyRootDirectory(@TempDir(cleanup =
ALWAYS) File tempDir) {
+File rootDirectory = new File(tempDir, "rootDirectory");
+
+rootDirectory.mkdir();
+
+WorkingDirectoryUtils.purgeIncompleteUnpackedNars(rootDirectory);
+
+assertTrue(rootDirectory.exists());
+}
+
+@Test
+public void
testPurgeUnpackedNarsRootDirectoryWithFilesOnly(@TempDir(cleanup = ALWAYS) File
tempDir) throws IOException {
+File rootDirectory = new File(tempDir, "rootDirectory");
+File directoryContent1 = new File(rootDirectory, "file1");
+File directoryContent2 = new File(rootDirectory, "file2");
+
+rootDirectory.mkdir();
+directoryContent1.createNewFile();
+directoryContent2.createNewFile();
+
+WorkingDirectoryUtils.purgeIncompleteUnpackedNars(rootDirectory);
+
+assertTrue(rootDirectory.exists() && directoryContent1.exists() &&
directoryContent2.exists());
+}
+
+@Test
+public void testPurgeUnpackedNars(@TempDir(cleanup = ALWAYS) File tempDir)
throws IOException {
+File
[nifi] branch support/nifi-1.x updated (4d00f8e156 -> 19153a7609)
This is an automated email from the ASF dual-hosted git repository. exceptionfactory pushed a change to branch support/nifi-1.x in repository https://gitbox.apache.org/repos/asf/nifi.git from 4d00f8e156 NIFI-12228: This closes #7882. Fixed issue with FlowFile Concucrrency that can occasionally bring in more data than it should. new 387c22b22c NIFI-12122 Fixed persistence of Parameter Context descriptions new 19153a7609 NIFI-12160 Kafka Connect: Check for NAR unpacking before starting The 2 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../kafka/connect/WorkingDirectoryUtilsTest.java | 222 + .../kafka/connect/StatelessKafkaConnectorUtil.java | 10 +- .../nifi/kafka/connect/WorkingDirectoryUtils.java | 100 ++ .../nifi/controller/flow/AbstractFlowManager.java | 5 +- .../StandardVersionedComponentSynchronizer.java| 6 +- ...StandardVersionedComponentSynchronizerTest.java | 10 +- .../apache/nifi/controller/flow/FlowManager.java | 5 +- .../serialization/VersionedFlowSynchronizer.java | 4 +- .../apache/nifi/controller/TestFlowController.java | 30 ++- .../conf/parameter-context-flow-description.json | 60 ++ .../web/dao/impl/StandardParameterContextDAO.java | 2 +- 11 files changed, 436 insertions(+), 18 deletions(-) create mode 100644 nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java create mode 100644 nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtils.java create mode 100644 nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/parameter-context-flow-description.json
[nifi] 01/02: NIFI-12122 Fixed persistence of Parameter Context descriptions
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 387c22b22c8583082d15acd64868c0e3bd1454e3
Author: Zackeus Bengtsson
AuthorDate: Mon Sep 25 14:24:10 2023 +0200
NIFI-12122 Fixed persistence of Parameter Context descriptions
Fixed bug where parameter context descriptions were not loaded on NiFi
startup and overwritten as empty
This closes: #7787
Signed-off-by: David Handermann
(cherry picked from commit 0a47157640c1502a0379a7f66d764d2d7d344872)
---
.../nifi/controller/flow/AbstractFlowManager.java | 5 +-
.../StandardVersionedComponentSynchronizer.java| 6 ++-
...StandardVersionedComponentSynchronizerTest.java | 10 ++--
.../apache/nifi/controller/flow/FlowManager.java | 5 +-
.../serialization/VersionedFlowSynchronizer.java | 4 +-
.../apache/nifi/controller/TestFlowController.java | 30 +--
.../conf/parameter-context-flow-description.json | 60 ++
.../web/dao/impl/StandardParameterContextDAO.java | 2 +-
8 files changed, 107 insertions(+), 15 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
index 3b0d7681ea..e21b6621b5 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
@@ -507,8 +507,8 @@ public abstract class AbstractFlowManager implements
FlowManager {
}
@Override
-public ParameterContext createParameterContext(final String id, final
String name, final Map parameters,
- final List
inheritedContextIds,
+public ParameterContext createParameterContext(final String id, final
String name, final String description,
+ final Map parameters, final List inheritedContextIds,
final
ParameterProviderConfiguration parameterProviderConfiguration) {
final boolean namingConflict =
parameterContextManager.getParameterContexts().stream()
.anyMatch(paramContext -> paramContext.getName().equals(name));
@@ -527,6 +527,7 @@ public abstract class AbstractFlowManager implements
FlowManager {
.parameterProviderConfiguration(parameterProviderConfiguration)
.build();
parameterContext.setParameters(parameters);
+parameterContext.setDescription(description);
if (inheritedContextIds != null && !inheritedContextIds.isEmpty()) {
if (!withParameterContextResolution.get()) {
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
index 4993a415d4..6abc253506 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
@@ -2084,7 +2084,8 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
parameters.put(versionedParameter.getName(), parameter);
}
-return
context.getFlowManager().createParameterContext(parameterContextId,
versionedParameterContext.getName(), parameters, Collections.emptyList(), null);
+return
context.getFlowManager().createParameterContext(parameterContextId,
versionedParameterContext.getName(), versionedParameterContext.getDescription(),
+ parameters,
Collections.emptyList(), null);
}
private ParameterProviderConfiguration
getParameterProviderConfiguration(final VersionedParameterContext context) {
@@ -2107,7 +2108,8 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
final AtomicReference contextReference = new
AtomicReference<>();
context.getFlowManager().withParameterContextResolution(() -> {
-final ParameterContext created =
[nifi] 01/02: NIFI-12122 Fixed persistence of Parameter Context descriptions
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 0a47157640c1502a0379a7f66d764d2d7d344872
Author: Zackeus Bengtsson
AuthorDate: Mon Sep 25 14:24:10 2023 +0200
NIFI-12122 Fixed persistence of Parameter Context descriptions
Fixed bug where parameter context descriptions were not loaded on NiFi
startup and overwritten as empty
This closes: #7787
Signed-off-by: David Handermann
---
.../nifi/controller/flow/AbstractFlowManager.java | 5 +-
.../StandardVersionedComponentSynchronizer.java| 6 ++-
...StandardVersionedComponentSynchronizerTest.java | 10 ++--
.../apache/nifi/controller/flow/FlowManager.java | 5 +-
.../serialization/VersionedFlowSynchronizer.java | 4 +-
.../apache/nifi/controller/TestFlowController.java | 30 +--
.../conf/parameter-context-flow-description.json | 60 ++
.../web/dao/impl/StandardParameterContextDAO.java | 2 +-
8 files changed, 107 insertions(+), 15 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
index 4a8a714398..176a0d01cf 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/flow/AbstractFlowManager.java
@@ -594,8 +594,8 @@ public abstract class AbstractFlowManager implements
FlowManager {
}
@Override
-public ParameterContext createParameterContext(final String id, final
String name, final Map parameters,
- final List
inheritedContextIds,
+public ParameterContext createParameterContext(final String id, final
String name, final String description,
+ final Map parameters, final List inheritedContextIds,
final
ParameterProviderConfiguration parameterProviderConfiguration) {
final boolean namingConflict =
parameterContextManager.getParameterContexts().stream()
.anyMatch(paramContext -> paramContext.getName().equals(name));
@@ -614,6 +614,7 @@ public abstract class AbstractFlowManager implements
FlowManager {
.parameterProviderConfiguration(parameterProviderConfiguration)
.build();
parameterContext.setParameters(parameters);
+parameterContext.setDescription(description);
if (inheritedContextIds != null && !inheritedContextIds.isEmpty()) {
if (!withParameterContextResolution.get()) {
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
index df837ac17e..3a2a5f76fc 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/flow/synchronization/StandardVersionedComponentSynchronizer.java
@@ -1990,7 +1990,8 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
parameters.put(versionedParameter.getName(), parameter);
}
-return
context.getFlowManager().createParameterContext(parameterContextId,
versionedParameterContext.getName(), parameters, Collections.emptyList(), null);
+return
context.getFlowManager().createParameterContext(parameterContextId,
versionedParameterContext.getName(), versionedParameterContext.getDescription(),
+ parameters,
Collections.emptyList(), null);
}
private ParameterProviderConfiguration
getParameterProviderConfiguration(final VersionedParameterContext context) {
@@ -2013,7 +2014,8 @@ public class StandardVersionedComponentSynchronizer
implements VersionedComponen
final AtomicReference contextReference = new
AtomicReference<>();
context.getFlowManager().withParameterContextResolution(() -> {
-final ParameterContext created =
context.getFlowManager().createParameterContext(parameterContextId,
versionedParameterContext.getName(),
[nifi] branch main updated (3ae0eedee6 -> b2e3898e17)
This is an automated email from the ASF dual-hosted git repository. exceptionfactory pushed a change to branch main in repository https://gitbox.apache.org/repos/asf/nifi.git from 3ae0eedee6 NIFI-12124: This closes #7791. Added a new RenameRecordField processor. In testing, also noticed that the Descendant Wildcard operator (//*) and Descendant Field Path Operator (//name, for instance) did not properly account for array of records or map elements, so addressed those concerns. new 0a47157640 NIFI-12122 Fixed persistence of Parameter Context descriptions new b2e3898e17 NIFI-12160 Kafka Connect: Check for NAR unpacking before starting The 2 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: .../kafka/connect/WorkingDirectoryUtilsTest.java | 222 + .../kafka/connect/StatelessKafkaConnectorUtil.java | 10 +- .../nifi/kafka/connect/WorkingDirectoryUtils.java | 100 ++ .../nifi/controller/flow/AbstractFlowManager.java | 5 +- .../StandardVersionedComponentSynchronizer.java| 6 +- ...StandardVersionedComponentSynchronizerTest.java | 10 +- .../apache/nifi/controller/flow/FlowManager.java | 5 +- .../serialization/VersionedFlowSynchronizer.java | 4 +- .../apache/nifi/controller/TestFlowController.java | 30 ++- .../conf/parameter-context-flow-description.json | 60 ++ .../web/dao/impl/StandardParameterContextDAO.java | 2 +- 11 files changed, 436 insertions(+), 18 deletions(-) create mode 100644 nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java create mode 100644 nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtils.java create mode 100644 nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/parameter-context-flow-description.json
[nifi] 02/02: NIFI-12160 Kafka Connect: Check for NAR unpacking before starting
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit b2e3898e171b89e25b5e06f6ee849bb3edabbf9a
Author: Peter Gyori
AuthorDate: Tue Oct 3 14:42:29 2023 +0200
NIFI-12160 Kafka Connect: Check for NAR unpacking before starting
Check that required NAR files are unpacked completely before starting the
Kafka Connector
This closes #7832
Signed-off-by: David Handermann
---
.../kafka/connect/WorkingDirectoryUtilsTest.java | 222 +
.../kafka/connect/StatelessKafkaConnectorUtil.java | 10 +-
.../nifi/kafka/connect/WorkingDirectoryUtils.java | 100 ++
3 files changed, 329 insertions(+), 3 deletions(-)
diff --git
a/nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java
b/nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java
new file mode 100644
index 00..e98fcaabb4
--- /dev/null
+++
b/nifi-external/nifi-kafka-connect/nifi-kafka-connector-tests/src/test/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtilsTest.java
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.kafka.connect;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.io.CleanupMode.ALWAYS;
+
+public class WorkingDirectoryUtilsTest {
+
+@Test
+public void testDeleteNonexistentFile(@TempDir(cleanup = ALWAYS) File
tempDir) {
+File nonexistentFile = new File(tempDir, "testFile");
+
+WorkingDirectoryUtils.purgeDirectory(nonexistentFile);
+
+assertFalse(nonexistentFile.exists());
+}
+
+@Test
+public void testDeleteFlatFile(@TempDir(cleanup = ALWAYS) File tempDir)
throws IOException {
+File file = new File(tempDir, "testFile");
+file.createNewFile();
+
+WorkingDirectoryUtils.purgeDirectory(file);
+
+assertFalse(file.exists());
+}
+
+@Test
+public void testDeleteDirectoryWithContents(@TempDir(cleanup = ALWAYS)
File tempDir) throws IOException {
+File directory = new File(tempDir, "directory");
+File subDirectory = new File(directory, "subDirectory");
+File subDirectoryContent = new File(subDirectory,
"subDirectoryContent");
+File directoryContent = new File(directory, "directoryContent");
+
+directory.mkdir();
+subDirectory.mkdir();
+subDirectoryContent.createNewFile();
+directoryContent.createNewFile();
+
+WorkingDirectoryUtils.purgeDirectory(directory);
+
+assertFalse(directory.exists());
+}
+
+@Test
+public void testPurgeUnpackedNarsEmptyRootDirectory(@TempDir(cleanup =
ALWAYS) File tempDir) {
+File rootDirectory = new File(tempDir, "rootDirectory");
+
+rootDirectory.mkdir();
+
+WorkingDirectoryUtils.purgeIncompleteUnpackedNars(rootDirectory);
+
+assertTrue(rootDirectory.exists());
+}
+
+@Test
+public void
testPurgeUnpackedNarsRootDirectoryWithFilesOnly(@TempDir(cleanup = ALWAYS) File
tempDir) throws IOException {
+File rootDirectory = new File(tempDir, "rootDirectory");
+File directoryContent1 = new File(rootDirectory, "file1");
+File directoryContent2 = new File(rootDirectory, "file2");
+
+rootDirectory.mkdir();
+directoryContent1.createNewFile();
+directoryContent2.createNewFile();
+
+WorkingDirectoryUtils.purgeIncompleteUnpackedNars(rootDirectory);
+
+assertTrue(rootDirectory.exists() && directoryContent1.exists() &&
directoryContent2.exists());
+}
+
+@Test
+public void testPurgeUnpackedNars(@TempDir(cleanup = ALWAYS) File tempDir)
throws IOException {
+File rootDirectory = new File(tempDir, "rootDirectory");
+rootDirectory.mkdir();
+
[nifi] branch main updated: NIFI-12124: This closes #7791. Added a new RenameRecordField processor. In testing, also noticed that the Descendant Wildcard operator (//*) and Descendant Field Path Opera
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 3ae0eedee6 NIFI-12124: This closes #7791. Added a new
RenameRecordField processor. In testing, also noticed that the Descendant
Wildcard operator (//*) and Descendant Field Path Operator (//name, for
instance) did not properly account for array of records or map elements, so
addressed those concerns.
3ae0eedee6 is described below
commit 3ae0eedee6ae67c48dd59f68752fa728e4355c41
Author: Mark Payne
AuthorDate: Mon Sep 25 13:19:30 2023 -0400
NIFI-12124: This closes #7791. Added a new RenameRecordField processor. In
testing, also noticed that the Descendant Wildcard operator (//*) and
Descendant Field Path Operator (//name, for instance) did not properly account
for array of records or map elements, so addressed those concerns.
Signed-off-by: Joseph Witt
---
.github/workflows/ci-workflow.yml | 3 +
.../record/path/paths/DescendantFieldPath.java | 26 ++-
.../record/path/paths/WildcardDescendantPath.java | 26 ++-
.../org/apache/nifi/record/path/util/Filters.java | 38 +++-
.../apache/nifi/record/path/TestRecordPath.java| 76 ++--
.../nifi/serialization/SimpleRecordSchema.java | 70 +--
.../nifi/serialization/record/MapRecord.java | 28 +++
.../apache/nifi/serialization/record/Record.java | 14 +-
.../nifi/serialization/record/RecordSchema.java| 10 +
.../org/apache/nifi/accumulo/data/KeySchema.java | 14 +-
.../processors/standard/RenameRecordField.java | 209 +
.../services/org.apache.nifi.processor.Processor | 1 +
.../processors/standard/TestRenameRecordField.java | 182 ++
.../input/complex-person.json | 12 ++
.../TestRenameRecordField/input/simple-person.json | 5 +
.../output/testNestedPath.json | 12 ++
.../testRecursivelyReferencingAllFields.json | 12 ++
.../testRecursivelyReferencingFieldName.json | 12 ++
.../output/testReferencingFieldName.json | 5 +
.../output/testRenameArray.json| 12 ++
.../output/testRenameFieldStaticValue.json | 5 +
.../output/testRenameFieldUsingAttribute.json | 5 +
.../output/testRenameMultipleFields.json | 5 +
23 files changed, 740 insertions(+), 42 deletions(-)
diff --git a/.github/workflows/ci-workflow.yml
b/.github/workflows/ci-workflow.yml
index 3f9ee8a7f5..439fdff8b6 100644
--- a/.github/workflows/ci-workflow.yml
+++ b/.github/workflows/ci-workflow.yml
@@ -172,6 +172,7 @@ jobs:
if: failure()
- name: Post Disk Usage
run: df
+if: ${{ always() }}
macos-build-jp:
timeout-minutes: 150
@@ -235,6 +236,7 @@ jobs:
if: failure()
- name: Post Disk Usage
run: df
+if: ${{ always() }}
windows-build:
timeout-minutes: 150
@@ -300,3 +302,4 @@ jobs:
if: failure()
- name: Post Disk Usage
run: df
+if: ${{ always() }}
diff --git
a/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/DescendantFieldPath.java
b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/DescendantFieldPath.java
index 8946bc102c..60852f16ac 100644
---
a/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/DescendantFieldPath.java
+++
b/nifi-commons/nifi-record-path/src/main/java/org/apache/nifi/record/path/paths/DescendantFieldPath.java
@@ -20,14 +20,18 @@ package org.apache.nifi.record.path.paths;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.stream.Stream;
-
import org.apache.nifi.record.path.FieldValue;
+import org.apache.nifi.record.path.MapEntryFieldValue;
import org.apache.nifi.record.path.RecordPathEvaluationContext;
import org.apache.nifi.record.path.StandardFieldValue;
import org.apache.nifi.record.path.util.Filters;
+import org.apache.nifi.serialization.record.DataType;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordField;
+import org.apache.nifi.serialization.record.RecordFieldType;
+import org.apache.nifi.serialization.record.type.MapDataType;
public class DescendantFieldPath extends RecordPathSegment {
private final String descendantName;
@@ -74,6 +78,26 @@ public class DescendantFieldPath extends RecordPathSegment {
if (Filters.isRecord(childField.getDataType(), recordValue)) {
final FieldValue childFieldValue = new
StandardFieldValue(recordValue, childField, fieldValue);
matchingValues.addAll(findDescendants(childFieldValue));
+} else if (Filters.isRecordArray(childField.getDataType(),
recordValue)) {
+
[nifi] branch main updated: NIFI-12221: This closes #7876. Be more lenient about which Disconnection Codes we allow a node to be reconnected to a cluster vs. when we notify the node to disconnect agai
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new f4ae292a45 NIFI-12221: This closes #7876. Be more lenient about which
Disconnection Codes we allow a node to be reconnected to a cluster vs. when we
notify the node to disconnect again. Also updated the timeout for OffloadIT
because it occasionally times ou out while running properly.
f4ae292a45 is described below
commit f4ae292a457638d3226fb0491e5186fa52ae8518
Author: Mark Payne
AuthorDate: Thu Oct 12 16:43:21 2023 -0400
NIFI-12221: This closes #7876. Be more lenient about which Disconnection
Codes we allow a node to be reconnected to a cluster vs. when we notify the
node to disconnect again. Also updated the timeout for OffloadIT because it
occasionally times ou out while running properly.
Signed-off-by: Joseph Witt
---
.../coordination/heartbeat/AbstractHeartbeatMonitor.java | 12 +++-
.../org/apache/nifi/tests/system/clustering/OffloadIT.java | 5 +
2 files changed, 12 insertions(+), 5 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
index c8a77ed2f0..549b030804 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
@@ -254,20 +254,22 @@ public abstract class AbstractHeartbeatMonitor implements
HeartbeatMonitor {
case LACK_OF_HEARTBEAT:
case UNABLE_TO_COMMUNICATE:
case NOT_YET_CONNECTED:
-case STARTUP_FAILURE: {
+case MISMATCHED_FLOWS:
+case MISSING_BUNDLE:
+case NODE_SHUTDOWN:
+case FAILED_TO_SERVICE_REQUEST:
+case STARTUP_FAILURE:
clusterCoordinator.reportEvent(nodeId, Severity.INFO,
"Received heartbeat from node previously "
+ "disconnected due to " + disconnectionCode + ".
Issuing reconnection request.");
clusterCoordinator.requestNodeConnect(nodeId, null);
break;
-}
-default: {
+default:
// disconnected nodes should not heartbeat, so we need to
issue a disconnection request.
-logger.info("Ignoring received heartbeat from disconnected
node " + nodeId + ". Issuing disconnection request.");
+logger.info("Ignoring received heartbeat from disconnected
node {}. Node was disconnected due to [{}]. Issuing disconnection request.",
nodeId, disconnectionCode);
clusterCoordinator.requestNodeDisconnect(nodeId,
disconnectionCode, connectionStatus.getReason());
removeHeartbeat(nodeId);
break;
-}
}
return;
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
index 818bd0cb23..f2e2266d9d 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/clustering/OffloadIT.java
@@ -25,11 +25,13 @@ import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
import org.apache.nifi.web.api.entity.ConnectionEntity;
import org.apache.nifi.web.api.entity.ProcessorEntity;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collections;
+import java.util.concurrent.TimeUnit;
public class OffloadIT extends NiFiSystemIT {
private static final Logger logger =
LoggerFactory.getLogger(OffloadIT.class);
@@ -40,6 +42,9 @@ public class OffloadIT extends NiFiSystemIT {
}
@Test
+@Timeout(value = 10, unit = TimeUnit.MINUTES)
+// Test to ensure that node can be offloaded, reconnected, offloaded
several times. This test typically takes only about 1-2 minutes
+// but can occasionally take 5-6 minutes on Github Actions so we set the
timeout to 10 minutes to allow for these occasions
public void
[nifi] branch support/nifi-1.x updated: NIFI-12228: This closes #7882. Fixed issue with FlowFile Concucrrency that can occasionally bring in more data than it should.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new 4d00f8e156 NIFI-12228: This closes #7882. Fixed issue with FlowFile
Concucrrency that can occasionally bring in more data than it should.
4d00f8e156 is described below
commit 4d00f8e1563ffec40d8d601ac2b678ad0414bb5b
Author: Mark Payne
AuthorDate: Fri Oct 13 11:03:11 2023 -0400
NIFI-12228: This closes #7882. Fixed issue with FlowFile Concucrrency that
can occasionally bring in more data than it should.
Signed-off-by: Joseph Witt
---
.../nifi/groups/SingleConcurrencyFlowFileGate.java | 17 ++
.../org/apache/nifi/groups/StandardDataValve.java | 27 ++
2 files changed, 34 insertions(+), 10 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
index 2282636b57..c006cdb659 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
@@ -36,6 +36,16 @@ public class SingleConcurrencyFlowFileGate implements
FlowFileGate {
return false;
}
+// We need to try to open flow into the Port's group. To do this, we
need to get the data valve for the parent group,
+// as it is responsible for data flowing into and out of its children.
+final ProcessGroup dataValveGroup = port.getProcessGroup().getParent();
+final DataValve dataValve = dataValveGroup.getDataValve();
+final boolean openFlowIntoGroup =
dataValve.tryOpenFlowIntoGroup(port.getProcessGroup());
+if (!openFlowIntoGroup) {
+claimed.set(false);
+return false;
+}
+
// The claim is now held by this thread. Check if the ProcessGroup is
empty.
final boolean empty = !port.getProcessGroup().isDataQueued();
if (empty) {
@@ -43,6 +53,9 @@ public class SingleConcurrencyFlowFileGate implements
FlowFileGate {
return true;
}
+// We have already opened flow into group, so now we must close it,
since we are not allowing flow in
+dataValve.closeFlowIntoGroup(port.getProcessGroup());
+
// Process Group was not empty, so we cannot allow any more FlowFiles
through. Reset claimed to false and return false,
// indicating that the caller did not obtain the claim.
claimed.set(false);
@@ -52,5 +65,9 @@ public class SingleConcurrencyFlowFileGate implements
FlowFileGate {
@Override
public void releaseClaim(final Port port) {
claimed.set(false);
+
+final ProcessGroup dataValveGroup = port.getProcessGroup().getParent();
+final DataValve dataValve = dataValveGroup.getDataValve();
+dataValve.closeFlowIntoGroup(port.getProcessGroup());
}
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardDataValve.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardDataValve.java
index df5658638f..5c1c1ee5ec 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardDataValve.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardDataValve.java
@@ -84,10 +84,15 @@ public class StandardDataValve implements DataValve {
if (destinationGroup.isDataQueued()) {
// If the destination group already has data queued up, and the
valve is not already open, do not allow data to
// flow into the group. If we did, we would end up mixing together
two different batches of data.
-logger.debug("Will not allow data to flow into {} because valve is
not already open and the Process Group has data queued", destinationGroup);
+logger.trace("Will not allow data to flow into {} because valve is
not already open and the Process Group has data queued", destinationGroup);
return "Process Group already has data queued and valve is not
already allowing data into group";
}
+if (destinationGroup.getFlowFileOutboundPolicy() ==
FlowFileOutboundPolicy.BATCH_OUTPUT &&
[nifi] branch main updated: NIFI-12228: This closes #7881. Fixed issue with FlowFile Concucrrency that can occasionally bring in more data than it should. Code cleanup, fixing logback to avoid INFO-le
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 0eabbcdf19 NIFI-12228: This closes #7881. Fixed issue with FlowFile
Concucrrency that can occasionally bring in more data than it should. Code
cleanup, fixing logback to avoid INFO-level stack trace from xodus
0eabbcdf19 is described below
commit 0eabbcdf19dfa9de7085b4b173ac3d3260f9d3a4
Author: Mark Payne
AuthorDate: Fri Oct 13 11:03:11 2023 -0400
NIFI-12228: This closes #7881. Fixed issue with FlowFile Concucrrency that
can occasionally bring in more data than it should.
Code cleanup, fixing logback to avoid INFO-level stack trace from xodus
Signed-off-by: Joseph Witt
---
.../org/apache/nifi/connectable/LocalPort.java | 12 +++---
.../nifi/groups/SingleConcurrencyFlowFileGate.java | 17 ++
.../org/apache/nifi/groups/StandardDataValve.java | 27 ++
.../resources/conf/clustered/node1/logback.xml | 3 +++
.../resources/conf/clustered/node2/logback.xml | 3 +++
.../src/test/resources/conf/default/logback.xml| 4 +++-
.../src/test/resources/conf/pythonic/logback.xml | 3 +++
7 files changed, 49 insertions(+), 20 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/LocalPort.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/LocalPort.java
index 86366f6eb8..53ffc01937 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/LocalPort.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/connectable/LocalPort.java
@@ -186,15 +186,9 @@ public class LocalPort extends AbstractPort {
final FlowFileConcurrency flowFileConcurrency =
getProcessGroup().getFlowFileConcurrency();
switch (flowFileConcurrency) {
-case UNBOUNDED:
-transferUnboundedConcurrency(context, session);
-break;
-case SINGLE_FLOWFILE_PER_NODE:
-transferSingleFlowFile(session);
-break;
-case SINGLE_BATCH_PER_NODE:
-transferInputBatch(session);
-break;
+case UNBOUNDED -> transferUnboundedConcurrency(context,
session);
+case SINGLE_FLOWFILE_PER_NODE ->
transferSingleFlowFile(session);
+case SINGLE_BATCH_PER_NODE -> transferInputBatch(session);
}
} finally {
flowFileGate.releaseClaim(this);
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
index 2282636b57..c006cdb659 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/groups/SingleConcurrencyFlowFileGate.java
@@ -36,6 +36,16 @@ public class SingleConcurrencyFlowFileGate implements
FlowFileGate {
return false;
}
+// We need to try to open flow into the Port's group. To do this, we
need to get the data valve for the parent group,
+// as it is responsible for data flowing into and out of its children.
+final ProcessGroup dataValveGroup = port.getProcessGroup().getParent();
+final DataValve dataValve = dataValveGroup.getDataValve();
+final boolean openFlowIntoGroup =
dataValve.tryOpenFlowIntoGroup(port.getProcessGroup());
+if (!openFlowIntoGroup) {
+claimed.set(false);
+return false;
+}
+
// The claim is now held by this thread. Check if the ProcessGroup is
empty.
final boolean empty = !port.getProcessGroup().isDataQueued();
if (empty) {
@@ -43,6 +53,9 @@ public class SingleConcurrencyFlowFileGate implements
FlowFileGate {
return true;
}
+// We have already opened flow into group, so now we must close it,
since we are not allowing flow in
+dataValve.closeFlowIntoGroup(port.getProcessGroup());
+
// Process Group was not empty, so we cannot allow any more FlowFiles
through. Reset claimed to false and return false,
// indicating that the caller did not obtain the claim.
claimed.set(false);
@@ -52,5
[nifi] branch main updated (22ad7d542d -> 96eb1d825a)
This is an automated email from the ASF dual-hosted git repository. markap14 pushed a change to branch main in repository https://gitbox.apache.org/repos/asf/nifi.git from 22ad7d542d NIFI-12206 Refactor Flow History using JetBrains Xodus (#7870) add 96eb1d825a NIFI-1 Protect against missing parameter context when syncing a PG in component synchronizer (#7877) No new revisions were added by this update. Summary of changes: .../StandardVersionedComponentSynchronizer.java| 41 - ...StandardVersionedComponentSynchronizerTest.java | 53 ++ 2 files changed, 71 insertions(+), 23 deletions(-)
