You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2023/10/13 23:40:18 UTC
[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
Posted by ex...@apache.org.
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 <gi...@zackeus.no>
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 <ex...@apache.org>
(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<String, Parameter> parameters,
- final List<String> inheritedContextIds,
+ public ParameterContext createParameterContext(final String id, final String name, final String description,
+ final Map<String, Parameter> parameters, final List<String> 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<ParameterContext> contextReference = new AtomicReference<>();
context.getFlowManager().withParameterContextResolution(() -> {
- final ParameterContext created = context.getFlowManager().createParameterContext(parameterContextId, versionedParameterContext.getName(), parameters, parameterContextRefs,
+ final ParameterContext created = context.getFlowManager().createParameterContext(parameterContextId, versionedParameterContext.getName(),
+ versionedParameterContext.getDescription(), parameters, parameterContextRefs,
getParameterProviderConfiguration(versionedParameterContext));
contextReference.set(created);
});
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 2e356828c1..4f4a021291 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
@@ -176,10 +176,14 @@ public class StandardVersionedComponentSynchronizerTest {
.parameterReferenceManager(parameterReferenceManager)
.build();
- final Map<String, Parameter> parameterMap = invocation.getArgument(2, Map.class);
+ final String description = invocation.getArgument(2, String.class);
+ parameterContext.setDescription(description);
+
+ final Map<String, Parameter> parameterMap = invocation.getArgument(3, Map.class);
parameterContext.setParameters(parameterMap);
- final List<String> inheritedContextIds = invocation.getArgument(3, List.class);
+
+ final List<String> inheritedContextIds = invocation.getArgument(4, List.class);
final List<ParameterContext> inheritedContexts = inheritedContextIds.stream()
.map(parameterContextManager::getParameterContext)
.collect(Collectors.toList());
@@ -188,7 +192,7 @@ public class StandardVersionedComponentSynchronizerTest {
parameterContextManager.addParameterContext(parameterContext);
return parameterContext;
- }).when(flowManager).createParameterContext(anyString(), anyString(), anyMap(), anyList(), or(any(ParameterProviderConfiguration.class), isNull()));
+ }).when(flowManager).createParameterContext(anyString(), anyString(), anyString(), anyMap(), anyList(), or(any(ParameterProviderConfiguration.class), isNull()));
final VersionedFlowSynchronizationContext context = new VersionedFlowSynchronizationContext.Builder()
.componentIdGenerator(componentIdGenerator)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/flow/FlowManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/flow/FlowManager.java
index c33a80d228..61b1fec1e3 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/flow/FlowManager.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/flow/FlowManager.java
@@ -359,6 +359,7 @@ public interface FlowManager extends ParameterProviderLookup {
*
* @param id The unique id
* @param name The ParameterContext name
+ * @param description The ParameterContext description
* @param parameters The Parameters
* @param inheritedContextIds The identifiers of any Parameter Contexts that the newly created Parameter Context should inherit from. The order of the identifiers in the List determines the
* order in which parameters with conflicting names are resolved. I.e., the Parameter Context whose ID comes first in the List is preferred.
@@ -367,8 +368,8 @@ public interface FlowManager extends ParameterProviderLookup {
* @throws IllegalStateException If <code>parameterContexts</code> is not empty and this method is called without being wrapped
* by {@link FlowManager#withParameterContextResolution(Runnable)}
*/
- ParameterContext createParameterContext(String id, String name, Map<String, Parameter> parameters, List<String> inheritedContextIds,
- ParameterProviderConfiguration parameterProviderConfiguration);
+ ParameterContext createParameterContext(String id, String name, String description, Map<String, Parameter> parameters,
+ List<String> inheritedContextIds, ParameterProviderConfiguration parameterProviderConfiguration);
/**
* Performs the given ParameterContext-related action, and then resolves all inherited ParameterContext references.
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
index 74723f4a92..cf005441bd 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
@@ -729,8 +729,8 @@ public class VersionedFlowSynchronizer implements FlowSynchronizer {
parameterProviderConfiguration = new StandardParameterProviderConfiguration(versionedParameterContext.getParameterProvider(),
versionedParameterContext.getParameterGroupName(), versionedParameterContext.isSynchronized());
}
- flowManager.createParameterContext(versionedParameterContext.getInstanceIdentifier(), versionedParameterContext.getName(), parameters, referenceIds,
- parameterProviderConfiguration);
+ flowManager.createParameterContext(versionedParameterContext.getInstanceIdentifier(), versionedParameterContext.getName(), versionedParameterContext.getDescription(),
+ parameters, referenceIds, parameterProviderConfiguration);
logger.info("Added Parameter Context {}", versionedParameterContext.getName());
}
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 069b379e8b..fc0e4b7bd1 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
@@ -120,6 +120,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -485,7 +486,7 @@ public class TestFlowController {
parameters.put("param", new Parameter(new ParameterDescriptor.Builder().name("param").build(), "value"));
// No problem since there are no inherited parameter contexts
- controller.getFlowManager().createParameterContext("id", "name", parameters, Collections.emptyList(), null);
+ controller.getFlowManager().createParameterContext("id", "name", "description", parameters, Collections.emptyList(), null);
final ParameterContext existingParameterContext = controller.getFlowManager().getParameterContextManager().getParameterContext("context");
final ParameterContextReferenceDTO dto = new ParameterContextReferenceDTO();
@@ -494,17 +495,40 @@ public class TestFlowController {
// This is not wrapped in FlowManager#withParameterContextResolution(Runnable), so it will throw an exception
assertThrows(IllegalStateException.class, () ->
- controller.getFlowManager().createParameterContext("id", "name", parameters, Collections.singletonList(existingParameterContext.getIdentifier()), null));
+ controller.getFlowManager().createParameterContext("id", "name", "description", parameters, Collections.singletonList(existingParameterContext.getIdentifier()), null));
// Instead, this is how it should be called
controller.getFlowManager().withParameterContextResolution(() -> controller
- .getFlowManager().createParameterContext("id2", "name2", parameters, Collections.singletonList(existingParameterContext.getIdentifier()), null));
+ .getFlowManager().createParameterContext("id2", "name2", "description2", parameters, Collections.singletonList(existingParameterContext.getIdentifier()), null));
} finally {
purgeFlow();
}
}
+ @Test
+ public void testCreateParameterContextLoadsDescription() throws IOException {
+ final String authFingerprint = authorizer.getFingerprint();
+ final File flowFile = new File("src/test/resources/conf/parameter-context-flow-description.json");
+ 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());
+
+ try {
+ controller.synchronize(flowSynchronizer, proposedDataFlow, mock(FlowService.class), BundleUpdateStrategy.IGNORE_BUNDLE);
+ controller.initializeFlow();
+
+ ParameterContext parameterContext = controller.getFlowManager().getParameterContextManager().getParameterContext("context");
+ assertNotNull(parameterContext);
+ assertNull(parameterContext.getDescription());
+
+ ParameterContext parameterContext2 = controller.getFlowManager().getParameterContextManager().getParameterContext("context2");
+ assertNotNull(parameterContext2);
+ assertEquals("description", parameterContext2.getDescription());
+ } finally {
+ purgeFlow();
+ }
+ }
+
private void purgeFlow() {
final ProcessGroup processGroup = controller.getFlowManager().getRootGroup();
for (final ProcessorNode procNode : processGroup.getProcessors()) {
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/parameter-context-flow-description.json b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/parameter-context-flow-description.json
new file mode 100644
index 0000000000..358172701a
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/parameter-context-flow-description.json
@@ -0,0 +1,60 @@
+{
+ "encodingVersion": {
+ "majorVersion": 2,
+ "minorVersion": 0
+ },
+ "maxTimerDrivenThreadCount": 10,
+ "registries": [],
+ "parameterContexts": [
+ {
+ "identifier": "e5033705-0219-3aa5-88d4-542cd7f1c13d",
+ "instanceIdentifier": "context",
+ "name": "context",
+ "parameters": [],
+ "inheritedParameterContexts": [],
+ "componentType": "PARAMETER_CONTEXT"
+ },
+ {
+ "identifier": "8304c86e-2662-3f0b-9b54-153015055f25",
+ "instanceIdentifier": "context2",
+ "name": "context2",
+ "parameters": [],
+ "inheritedParameterContexts": [],
+ "description": "description",
+ "componentType": "PARAMETER_CONTEXT"
+ }
+ ],
+ "parameterProviders": [],
+ "controllerServices": [],
+ "reportingTasks": [],
+ "flowAnalysisRules": [],
+ "rootGroup": {
+ "identifier": "55cf3437-cecb-3f77-806a-b742995a79b9",
+ "instanceIdentifier": "e6bf1066-018a-1000-97f3-ca4c6fe9e05c",
+ "name": "NiFi Flow",
+ "comments": "",
+ "position": {
+ "x": 0.0,
+ "y": 0.0
+ },
+ "processGroups": [],
+ "remoteProcessGroups": [],
+ "processors": [],
+ "inputPorts": [],
+ "outputPorts": [],
+ "connections": [],
+ "labels": [],
+ "funnels": [],
+ "controllerServices": [],
+ "defaultFlowFileExpiration": "0 sec",
+ "defaultBackPressureObjectThreshold": 10000,
+ "defaultBackPressureDataSizeThreshold": "1 GB",
+ "scheduledState": "ENABLED",
+ "executionEngine": "INHERITED",
+ "maxConcurrentTasks": 1,
+ "statelessFlowTimeout": "1 min",
+ "componentType": "PROCESS_GROUP",
+ "flowFileOutboundPolicy": "STREAM_WHEN_AVAILABLE",
+ "flowFileConcurrency": "UNBOUNDED"
+ }
+}
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java
index 6504fd943d..dc471d67bd 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java
@@ -99,7 +99,7 @@ public class StandardParameterContextDAO implements ParameterContextDAO {
final ParameterProviderConfiguration parameterProviderConfiguration = createParameterProviderConfiguration(parameterProviderConfigurationEntity);
final ParameterContext parameterContext = flowManager.createParameterContext(parameterContextDto.getId(), parameterContextDto.getName(),
- parameters, referencedIds, parameterProviderConfiguration);
+ parameterContextDto.getDescription(), parameters, referencedIds, parameterProviderConfiguration);
if (parameterContextDto.getDescription() != null) {
parameterContext.setDescription(parameterContextDto.getDescription());
}
[nifi] 02/02: NIFI-12160 Kafka Connect: Check for NAR unpacking before starting
Posted by ex...@apache.org.
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 <pg...@apache.org>
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 <ex...@apache.org>
(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 0000000000..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();
+ TestDirectoryStructure testDirectoryStructure = new TestDirectoryStructure(rootDirectory);
+
+ WorkingDirectoryUtils.purgeIncompleteUnpackedNars(testDirectoryStructure.getRootDirectory());
+
+ assertTrue(testDirectoryStructure.isConsistent());
+ }
+
+ @Test
+ public void testWorkingDirectoryIntegrityRestored(@TempDir(cleanup = ALWAYS) File tempDir) throws IOException {
+ /*
+ workingDirectory
+ - nar
+ - extensions
+ - *TestDirectoryStructure*
+ - narDirectory
+ - narFile
+ - extensions
+ - *TestDirectoryStructure*
+ - additionalDirectory
+ - workingDirectoryFile
+ */
+ File workingDirectory = new File(tempDir, "workingDirectory");
+ File nar = new File(workingDirectory, "nar");
+ File narExtensions = new File(nar, "extensions");
+ File narDirectory = new File(nar, "narDirectory");
+ File narFile = new File(nar, "narFile");
+ File extensions = new File(workingDirectory, "extensions");
+ File additionalDirectory = new File(workingDirectory, "additionalDirectory");
+ File workingDirectoryFile = new File(workingDirectory, "workingDirectoryFile");
+
+ workingDirectory.mkdir();
+ nar.mkdir();
+ narExtensions.mkdir();
+ narDirectory.mkdir();
+ narFile.createNewFile();
+ extensions.mkdir();
+ additionalDirectory.mkdir();
+ workingDirectoryFile.createNewFile();
+
+ TestDirectoryStructure narExtensionsStructure = new TestDirectoryStructure(narExtensions);
+ TestDirectoryStructure extensionsStructure = new TestDirectoryStructure(extensions);
+
+ WorkingDirectoryUtils.reconcileWorkingDirectory(workingDirectory);
+
+ assertTrue(workingDirectory.exists()
+ && nar.exists()
+ && narExtensionsStructure.isConsistent()
+ && narDirectory.exists()
+ && narFile.exists()
+ && extensionsStructure.isConsistent()
+ && additionalDirectory.exists()
+ && workingDirectoryFile.exists()
+ );
+ }
+
+ private class TestDirectoryStructure {
+ /*
+ rootDirectory
+ - subDirectory1-nar-unpacked
+ - subDirectory1File1
+ - nar-digest
+ - subDirectory2
+ - subDirectory2File1
+ - subDirectory3-nar-unpacked
+ - subDirectory3Dir1
+ - subDirectory3Dir1File1
+ - subDirectory3File1
+ - fileInRoot
+ */
+ File rootDirectory;
+ File subDirectory1;
+ File subDirectory2;
+ File subDirectory3;
+ File fileInRoot;
+ File subDirectory1File1;
+ File subDirectory1File2;
+ File subDirectory2File1;
+ File subDirectory3Dir1;
+ File subDirectory3File1;
+ File subDirectory3Dir1File1;
+
+ public TestDirectoryStructure(final File rootDirectory) throws IOException {
+ this.rootDirectory = rootDirectory;
+ subDirectory1 = new File(rootDirectory, "subDirectory1-" + WorkingDirectoryUtils.NAR_UNPACKED_SUFFIX);
+ subDirectory2 = new File(rootDirectory, "subDirector2");
+ subDirectory3 = new File(rootDirectory, "subDirector3-" + WorkingDirectoryUtils.NAR_UNPACKED_SUFFIX);
+ fileInRoot = new File(rootDirectory, "fileInRoot");
+ subDirectory1File1 = new File(subDirectory1, "subDirectory1File1");
+ subDirectory1File2 = new File(subDirectory1, WorkingDirectoryUtils.HASH_FILENAME);
+ subDirectory2File1 = new File(subDirectory2, "subDirectory2File1");
+ subDirectory3Dir1 = new File(subDirectory3, "subDirectory3Dir1");
+ subDirectory3File1 = new File(subDirectory3, "subDirectory3File1");
+ subDirectory3Dir1File1 = new File(subDirectory3Dir1, "subDirectory3Dir1File1");
+
+ subDirectory1.mkdir();
+ subDirectory2.mkdir();
+ subDirectory3.mkdir();
+ fileInRoot.createNewFile();
+ subDirectory1File1.createNewFile();
+ subDirectory1File2.createNewFile();
+ subDirectory2File1.createNewFile();
+ subDirectory3File1.createNewFile();
+ subDirectory3Dir1.mkdir();
+ subDirectory3Dir1File1.createNewFile();
+ }
+
+ public File getRootDirectory() {
+ return rootDirectory;
+ }
+
+ /**
+ * Checks if all directories ending in 'nar-unpacked' that have a file named 'nar-digest' within still exist,
+ * and the directory ending in 'nar-unpacked' without 'nar-digest' has been removed with all of its contents.
+ * @return true if the above is met.
+ */
+ public boolean isConsistent() {
+ return (rootDirectory.exists()
+ && subDirectory1.exists() && subDirectory1File1.exists() && subDirectory1File2.exists()
+ && subDirectory2.exists() && subDirectory2File1.exists()
+ && !(subDirectory3.exists() || subDirectory3Dir1.exists() || subDirectory3File1.exists() || subDirectory3Dir1File1.exists())
+ && fileInRoot.exists());
+ }
+ }
+
+}
diff --git a/nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/StatelessKafkaConnectorUtil.java b/nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/StatelessKafkaConnectorUtil.java
index dcc138c977..5554217f42 100644
--- a/nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/StatelessKafkaConnectorUtil.java
+++ b/nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/StatelessKafkaConnectorUtil.java
@@ -84,6 +84,7 @@ public class StatelessKafkaConnectorUtil {
config.setFlowDefinition(dataflowDefinitionProperties);
dataflowDefinitionProperties.put(StatelessNiFiCommonConfig.BOOTSTRAP_FLOW_NAME, dataflowName);
MDC.setContextMap(Collections.singletonMap("dataflow", dataflowName));
+ StatelessDataflow dataflow;
// Use a Write Lock to ensure that only a single thread is calling StatelessBootstrap.bootstrap().
// We do this because the bootstrap() method will expand all NAR files into the working directory.
@@ -91,13 +92,16 @@ public class StatelessKafkaConnectorUtil {
// unpacking NARs at the same time, as it could potentially result in the working directory becoming corrupted.
unpackNarLock.lock();
try {
+ WorkingDirectoryUtils.reconcileWorkingDirectory(engineConfiguration.getWorkingDirectory());
+
bootstrap = StatelessBootstrap.bootstrap(engineConfiguration, StatelessNiFiSourceTask.class.getClassLoader());
+
+ dataflowDefinition = bootstrap.parseDataflowDefinition(dataflowDefinitionProperties, parameterOverrides);
+ dataflow = bootstrap.createDataflow(dataflowDefinition);
} finally {
unpackNarLock.unlock();
}
-
- dataflowDefinition = bootstrap.parseDataflowDefinition(dataflowDefinitionProperties, parameterOverrides);
- return bootstrap.createDataflow(dataflowDefinition);
+ return dataflow;
} catch (final Exception e) {
throw new RuntimeException("Failed to bootstrap Stateless NiFi Engine", e);
}
diff --git a/nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtils.java b/nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtils.java
new file mode 100644
index 0000000000..1453267a30
--- /dev/null
+++ b/nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/WorkingDirectoryUtils.java
@@ -0,0 +1,100 @@
+/*
+ * 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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.Arrays;
+
+public class WorkingDirectoryUtils {
+
+ protected static final String NAR_UNPACKED_SUFFIX = "nar-unpacked";
+ protected static final String HASH_FILENAME = "nar-digest";
+ private static final Logger logger = LoggerFactory.getLogger(WorkingDirectoryUtils.class);
+
+ /**
+ * Goes through the nar/extensions and extensions directories within the working directory
+ * and deletes every directory whose name ends in "nar-unpacked" and does not have a
+ * "nar-digest" file in it.
+ * @param workingDirectory File object pointing to the working directory.
+ */
+ public static void reconcileWorkingDirectory(final File workingDirectory) {
+ purgeIncompleteUnpackedNars(new File(new File(workingDirectory, "nar"), "extensions"));
+ purgeIncompleteUnpackedNars(new File(workingDirectory, "extensions"));
+ }
+
+ /**
+ * Receives a directory as parameter and goes through every directory within it that ends in
+ * "nar-unpacked". If a directory ending in "nar-unpacked" does not have a file named
+ * "nar-digest" within it, it gets deleted with all of its contents.
+ * @param directory A File object pointing to the directory that is supposed to contain
+ * further directories whose name ends in "nar-unpacked".
+ */
+ public static void purgeIncompleteUnpackedNars(final File directory) {
+ final File[] unpackedDirs = directory.listFiles(file -> file.isDirectory() && file.getName().endsWith(NAR_UNPACKED_SUFFIX));
+ if (unpackedDirs == null || unpackedDirs.length == 0) {
+ logger.debug("Found no unpacked NARs in {}", directory);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Directory contains: {}", Arrays.deepToString(directory.listFiles()));
+ }
+ return;
+ }
+
+ for (final File unpackedDir : unpackedDirs) {
+ final File narHashFile = new File(unpackedDir, HASH_FILENAME);
+ if (narHashFile.exists()) {
+ logger.debug("Already successfully unpacked {}", unpackedDir);
+ } else {
+ purgeDirectory(unpackedDir);
+ }
+ }
+ }
+
+ /**
+ * Delete a directory with all of its contents.
+ * @param directory The directory to be deleted.
+ */
+ public static void purgeDirectory(final File directory) {
+ if (directory.exists()) {
+ deleteRecursively(directory);
+ logger.debug("Cleaned up {}", directory);
+ }
+ }
+
+ private static void deleteRecursively(final File fileOrDirectory) {
+ if (fileOrDirectory.isDirectory()) {
+ final File[] files = fileOrDirectory.listFiles();
+ if (files != null) {
+ for (final File file : files) {
+ deleteRecursively(file);
+ }
+ }
+ }
+ deleteQuietly(fileOrDirectory);
+ }
+
+ private static void deleteQuietly(final File file) {
+ final boolean deleted = file.delete();
+ if (!deleted) {
+ logger.debug("Failed to cleanup temporary file {}", file);
+ }
+ }
+
+}