You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by fe...@apache.org on 2025/05/28 06:51:50 UTC
(nifi) branch main updated: NIFI-14547 MiNiFi - Resolve asset references in flows coming from C2 server
This is an automated email from the ASF dual-hosted git repository.
ferdei 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 eabbd508a1 NIFI-14547 MiNiFi - Resolve asset references in flows coming from C2 server
eabbd508a1 is described below
commit eabbd508a1960536e76354ea06bac8785e081876
Author: Peter Kedvessy <pk...@gmail.com>
AuthorDate: Thu May 8 08:57:42 2025 +0200
NIFI-14547 MiNiFi - Resolve asset references in flows coming from C2 server
Signed-off-by: Ferenc Erdei <er...@gmail.com>
This closes #9922.
---
.../service/MiNiFiConfigurationChangeListener.java | 6 +-
.../minifi/commons/service/FlowEnrichService.java | 3 +-
...ava => FlowPropertyAssetReferenceResolver.java} | 14 +--
.../commons/service/FlowPropertyEncryptor.java | 3 +-
.../commons/service/StandardFlowEnrichService.java | 4 +-
...dFlowPropertyAssetReferenceResolverService.java | 87 +++++++++++++
.../service/StandardFlowPropertyEncryptor.java | 4 +-
.../service/StandardFlowEnrichServiceTest.java | 22 ++--
...wPropertyAssetReferenceResolverServiceTest.java | 136 +++++++++++++++++++++
.../service/StandardFlowPropertyEncryptorTest.java | 12 +-
.../apache/nifi/minifi/c2/C2NifiClientService.java | 17 ++-
.../DefaultUpdateConfigurationStrategy.java | 26 ++--
.../syncresource/FileResourceRepository.java | 10 ++
.../command/syncresource/ResourceRepository.java | 2 +
.../SyncResourcePropertyProvider.java} | 25 ++--
.../DefaultUpdateConfigurationStrategyTest.java | 38 +++---
.../syncresource/FileResourceRepositoryTest.java | 22 ++++
17 files changed, 349 insertions(+), 82 deletions(-)
diff --git a/minifi/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/service/MiNiFiConfigurationChangeListener.java b/minifi/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/service/MiNiFiConfigurationChangeListener.java
index 5925ac3b5e..58098c5e44 100644
--- a/minifi/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/service/MiNiFiConfigurationChangeListener.java
+++ b/minifi/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/service/MiNiFiConfigurationChangeListener.java
@@ -88,9 +88,9 @@ public class MiNiFiConfigurationChangeListener implements ConfigurationChangeLis
backup(currentRawFlowConfigFile, backupRawFlowConfigFile);
byte[] rawFlow = toByteArray(flowConfigInputStream);
- VersionedDataflow rawDataFlow = flowSerDeService.deserialize(rawFlow);
- VersionedDataflow enrichedFlow = flowEnrichService.enrichFlow(rawDataFlow);
- byte[] serializedEnrichedFlow = flowSerDeService.serialize(enrichedFlow);
+ VersionedDataflow dataFlow = flowSerDeService.deserialize(rawFlow);
+ flowEnrichService.enrichFlow(dataFlow);
+ byte[] serializedEnrichedFlow = flowSerDeService.serialize(dataFlow);
persist(serializedEnrichedFlow, currentFlowConfigFile, true);
restartInstance();
persist(rawFlow, currentRawFlowConfigFile, false);
diff --git a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java b/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java
index 5b3620803f..ce15b07830 100644
--- a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java
+++ b/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java
@@ -29,7 +29,6 @@ public interface FlowEnrichService {
* Responsible for enriching a VersionedDataflow instance
*
* @param versionedDataflow a VersionedDataflow instance
- * @return VersionedDataflow the enriched flow instance
*/
- VersionedDataflow enrichFlow(VersionedDataflow versionedDataflow);
+ void enrichFlow(VersionedDataflow versionedDataflow);
}
diff --git a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java b/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyAssetReferenceResolver.java
similarity index 64%
copy from minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java
copy to minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyAssetReferenceResolver.java
index 5b3620803f..34e1dcef94 100644
--- a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java
+++ b/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyAssetReferenceResolver.java
@@ -19,17 +19,11 @@ package org.apache.nifi.minifi.commons.service;
import org.apache.nifi.controller.flow.VersionedDataflow;
-/**
- * Defines interface methods used to implement a FlowEnrichService.
- * The purpose of a flow enrich service is to enrich a VersionedDataFlow with various additional components specific to the MiNiFi instance
- */
-public interface FlowEnrichService {
-
+public interface FlowPropertyAssetReferenceResolver {
/**
- * Responsible for enriching a VersionedDataflow instance
+ * Responsible for resolving asset reference properties in a VersionedDataflow instance
*
- * @param versionedDataflow a VersionedDataflow instance
- * @return VersionedDataflow the enriched flow instance
+ * @param flow a VersionedDataflow instance to resolve its asset reference properties
*/
- VersionedDataflow enrichFlow(VersionedDataflow versionedDataflow);
+ void resolveAssetReferenceProperties(VersionedDataflow flow);
}
diff --git a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyEncryptor.java b/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyEncryptor.java
index 661b4bd242..f8481f21a3 100644
--- a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyEncryptor.java
+++ b/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowPropertyEncryptor.java
@@ -29,7 +29,6 @@ public interface FlowPropertyEncryptor {
* Responsible for encrypting sensitive properties in a VersionedDataflow instance
*
* @param flow a VersionedDataflow instance to encrypt its sensitive properties
- * @return VersionedDataflow the flow instance with encrypted sensitive properties
*/
- VersionedDataflow encryptSensitiveProperties(VersionedDataflow flow);
+ void encryptSensitiveProperties(VersionedDataflow flow);
}
diff --git a/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichService.java b/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichService.java
index 9695e4305f..6e12e91311 100644
--- a/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichService.java
+++ b/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichService.java
@@ -97,7 +97,7 @@ public class StandardFlowEnrichService implements FlowEnrichService {
}
@Override
- public VersionedDataflow enrichFlow(VersionedDataflow versionedDataflow) {
+ public void enrichFlow(VersionedDataflow versionedDataflow) {
versionedDataflow.setReportingTasks(ofNullable(versionedDataflow.getReportingTasks()).orElseGet(ArrayList::new));
versionedDataflow.setRegistries(ofNullable(versionedDataflow.getRegistries()).orElseGet(ArrayList::new));
versionedDataflow.setControllerServices(ofNullable(versionedDataflow.getControllerServices()).orElseGet(ArrayList::new));
@@ -147,8 +147,6 @@ public class StandardFlowEnrichService implements FlowEnrichService {
Map<String, String> idToInstanceIdMap = createIdToInstanceIdMap(rootGroup);
setConnectableComponentsInstanceId(rootGroup, idToInstanceIdMap);
}
-
- return versionedDataflow;
}
private void createDefaultParameterContext(VersionedDataflow versionedDataflow) {
diff --git a/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverService.java b/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverService.java
new file mode 100644
index 0000000000..e056984f4c
--- /dev/null
+++ b/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverService.java
@@ -0,0 +1,87 @@
+/*
+ * 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.minifi.commons.service;
+
+import org.apache.nifi.controller.flow.VersionedDataflow;
+import org.apache.nifi.flow.VersionedConfigurableExtension;
+import org.apache.nifi.flow.VersionedProcessGroup;
+
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import static java.util.Optional.ofNullable;
+import static java.util.stream.Stream.concat;
+
+public class StandardFlowPropertyAssetReferenceResolverService implements FlowPropertyAssetReferenceResolver {
+
+ private static final String ASSET_REFERENCE_PREFIX = "@{asset-id:";
+ private static final String ASSET_REFERENCE_SUFFIX = "}";
+ private static final String EMPTY_STRING = "";
+
+ private final Function<String, Optional<Path>> assetPathResolver;
+
+ public StandardFlowPropertyAssetReferenceResolverService(Function<String, Optional<Path>> assetPathResolver) {
+ this.assetPathResolver = assetPathResolver;
+ }
+
+ @Override
+ public void resolveAssetReferenceProperties(VersionedDataflow flow) {
+ fetchFlowComponents(flow).forEach(component -> {
+ component.getProperties().entrySet().stream()
+ .filter(e -> isAssetReference(e.getValue()))
+ .forEach(entry -> entry.setValue(getAssetAbsolutePathOrThrowIllegalStateException(entry.getValue())));
+ });
+ }
+
+ private boolean isAssetReference(String value) {
+ return value != null
+ && value.startsWith(ASSET_REFERENCE_PREFIX)
+ && value.endsWith(ASSET_REFERENCE_SUFFIX);
+ }
+
+ private Stream<? extends VersionedConfigurableExtension> fetchFlowComponents(VersionedDataflow flow) {
+ return concat(
+ ofNullable(flow.getControllerServices()).orElse(List.of()).stream(),
+ fetchComponentsRecursively(flow.getRootGroup())
+ );
+ }
+
+ private Stream<? extends VersionedConfigurableExtension> fetchComponentsRecursively(VersionedProcessGroup processGroup) {
+ return concat(
+ Stream.of(
+ ofNullable(processGroup.getProcessors()).orElse(Set.of()),
+ ofNullable(processGroup.getControllerServices()).orElse(Set.of())
+ )
+ .flatMap(Set::stream),
+ ofNullable(processGroup.getProcessGroups()).orElse(Set.of()).stream()
+ .flatMap(this::fetchComponentsRecursively)
+ );
+ }
+
+ private String getAssetAbsolutePathOrThrowIllegalStateException(String assetReference) {
+ String resourceId = assetReference.replace(ASSET_REFERENCE_PREFIX, EMPTY_STRING)
+ .replace(ASSET_REFERENCE_SUFFIX, EMPTY_STRING);
+ return assetPathResolver.apply(resourceId)
+ .map(Path::toString)
+ .orElseThrow(() -> new IllegalStateException("Resource '" + resourceId + "' not found"));
+ }
+}
diff --git a/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptor.java b/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptor.java
index 9222a41c2b..7f2308959b 100644
--- a/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptor.java
+++ b/minifi/minifi-commons/minifi-commons-framework/src/main/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptor.java
@@ -55,7 +55,7 @@ public class StandardFlowPropertyEncryptor implements FlowPropertyEncryptor {
}
@Override
- public VersionedDataflow encryptSensitiveProperties(VersionedDataflow flow) {
+ public void encryptSensitiveProperties(VersionedDataflow flow) {
encryptParameterContextsProperties(flow);
Map<String, Set<String>> sensitivePropertiesByComponentType = Optional.of(flowProvidedSensitiveProperties(flow))
@@ -63,8 +63,6 @@ public class StandardFlowPropertyEncryptor implements FlowPropertyEncryptor {
.orElseGet(this::runtimeManifestSensitiveProperties);
encryptFlowComponentsProperties(flow, sensitivePropertiesByComponentType);
-
- return flow;
}
private void encryptParameterContextsProperties(VersionedDataflow flow) {
diff --git a/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichServiceTest.java b/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichServiceTest.java
index 2501e30aab..e9f6a8dfd6 100644
--- a/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichServiceTest.java
+++ b/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowEnrichServiceTest.java
@@ -61,10 +61,10 @@ public class StandardFlowEnrichServiceTest {
VersionedDataflow testFlow = loadDefaultFlow();
FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
- VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+ testFlowEnrichService.enrichFlow(testFlow);
byte[] testFlowBytes = flowToString(testFlow).getBytes(UTF_8);
- byte[] enrichedFlowBytes = flowToString(enrichedFlow).getBytes(UTF_8);
+ byte[] enrichedFlowBytes = flowToString(testFlow).getBytes(UTF_8);
assertArrayEquals(testFlowBytes, enrichedFlowBytes);
}
@@ -80,10 +80,10 @@ public class StandardFlowEnrichServiceTest {
uuid.when(UUID::randomUUID).thenReturn(expectedIdentifier);
FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
- VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+ testFlowEnrichService.enrichFlow(testFlow);
- assertEquals(expectedIdentifier.toString(), enrichedFlow.getRootGroup().getIdentifier());
- assertEquals(expectedIdentifier.toString(), enrichedFlow.getRootGroup().getInstanceIdentifier());
+ assertEquals(expectedIdentifier.toString(), testFlow.getRootGroup().getIdentifier());
+ assertEquals(expectedIdentifier.toString(), testFlow.getRootGroup().getInstanceIdentifier());
}
}
@@ -100,13 +100,13 @@ public class StandardFlowEnrichServiceTest {
));
FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
- VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+ testFlowEnrichService.enrichFlow(testFlow);
- assertEquals(1, enrichedFlow.getRootGroup().getControllerServices().size());
- VersionedControllerService sslControllerService = enrichedFlow.getRootGroup().getControllerServices().iterator().next();
+ assertEquals(1, testFlow.getRootGroup().getControllerServices().size());
+ VersionedControllerService sslControllerService = testFlow.getRootGroup().getControllerServices().iterator().next();
assertEquals(PARENT_SSL_CONTEXT_SERVICE_NAME, sslControllerService.getName());
assertEquals(StringUtils.EMPTY, sslControllerService.getBundle().getVersion());
- Set<VersionedProcessor> processors = enrichedFlow.getRootGroup().getProcessors();
+ Set<VersionedProcessor> processors = testFlow.getRootGroup().getProcessors();
assertEquals(2, processors.size());
assertTrue(
processors.stream()
@@ -132,9 +132,9 @@ public class StandardFlowEnrichServiceTest {
VersionedDataflow testFlow = loadDefaultFlow();
FlowEnrichService testFlowEnrichService = new StandardFlowEnrichService(new StandardReadableProperties(properties));
- VersionedDataflow enrichedFlow = testFlowEnrichService.enrichFlow(testFlow);
+ testFlowEnrichService.enrichFlow(testFlow);
- List<VersionedReportingTask> reportingTasks = enrichedFlow.getReportingTasks();
+ List<VersionedReportingTask> reportingTasks = testFlow.getReportingTasks();
assertEquals(1, reportingTasks.size());
VersionedReportingTask provenanceReportingTask = reportingTasks.get(0);
assertEquals(SITE_TO_SITE_PROVENANCE_REPORTING_TASK_NAME, provenanceReportingTask.getName());
diff --git a/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverServiceTest.java b/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverServiceTest.java
new file mode 100644
index 0000000000..0d68309441
--- /dev/null
+++ b/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyAssetReferenceResolverServiceTest.java
@@ -0,0 +1,136 @@
+/*
+ * 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.minifi.commons.service;
+
+import org.apache.nifi.controller.flow.VersionedDataflow;
+import org.apache.nifi.flow.VersionedControllerService;
+import org.apache.nifi.flow.VersionedProcessGroup;
+import org.apache.nifi.flow.VersionedProcessor;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+public class StandardFlowPropertyAssetReferenceResolverServiceTest {
+
+ private Map<String, String> processorProperties;
+ private Map<String, String> controllerServiceProperties;
+
+ @Mock
+ private Function<String, Optional<Path>> assetPathResolver;
+
+ @InjectMocks
+ private StandardFlowPropertyAssetReferenceResolverService victim;
+
+ @Test
+ public void testResolveAssetReferenceProperties() {
+ initProperties();
+ VersionedDataflow dataFlow = aVersionedDataflow();
+
+ when(assetPathResolver.apply("asset1")).thenReturn(Optional.of(Path.of("asset1Path")));
+ when(assetPathResolver.apply("asset2")).thenReturn(Optional.of(Path.of("asset2Path")));
+
+ victim.resolveAssetReferenceProperties(dataFlow);
+
+ verifyProperties();
+ }
+
+ @Test
+ public void testResolveNestedAssetReferenceProperties() {
+ initProperties();
+ VersionedDataflow dataFlow = aVersionedDataflowWithNestedProcessGroup();
+
+ when(assetPathResolver.apply("asset1")).thenReturn(Optional.of(Path.of("asset1Path")));
+ when(assetPathResolver.apply("asset2")).thenReturn(Optional.of(Path.of("asset2Path")));
+
+ victim.resolveAssetReferenceProperties(dataFlow);
+
+ verifyProperties();
+ }
+
+ @Test
+ public void testResolveAssetReferencePropertiesThrowIllegalStateException() {
+ initProperties();
+ VersionedDataflow dataFlow = aVersionedDataflow();
+
+ when(assetPathResolver.apply("asset1")).thenReturn(Optional.empty());
+
+ assertThrows(IllegalStateException.class, () -> {
+ victim.resolveAssetReferenceProperties(dataFlow);
+ });
+ }
+
+ private void initProperties() {
+ processorProperties = new HashMap<>();
+ processorProperties.put("assetReferenceProperty", "@{asset-id:asset1}");
+ processorProperties.put("notAssetReferenceProperty", "some value1");
+
+ controllerServiceProperties = new HashMap<>();
+ controllerServiceProperties.put("assetReferenceProperty", "@{asset-id:asset2}");
+ controllerServiceProperties.put("notAssetReferenceProperty", "some value2");
+ }
+
+ private void verifyProperties() {
+ assertEquals(processorProperties.get("assetReferenceProperty"), "asset1Path");
+ assertEquals(processorProperties.get("notAssetReferenceProperty"), "some value1");
+ assertEquals(controllerServiceProperties.get("assetReferenceProperty"), "asset2Path");
+ assertEquals(controllerServiceProperties.get("notAssetReferenceProperty"), "some value2");
+ }
+
+ private VersionedDataflow aVersionedDataflow() {
+ VersionedDataflow versionedDataflow = new VersionedDataflow();
+ versionedDataflow.setRootGroup(aVersionedProcessGroup());
+ return versionedDataflow;
+ }
+
+ private VersionedDataflow aVersionedDataflowWithNestedProcessGroup() {
+ VersionedDataflow versionedDataflow = new VersionedDataflow();
+ VersionedProcessGroup versionedProcessGroup = new VersionedProcessGroup();
+ versionedProcessGroup.setProcessGroups(Set.of(aVersionedProcessGroup()));
+ versionedDataflow.setRootGroup(versionedProcessGroup);
+
+ return versionedDataflow;
+ }
+
+ private VersionedProcessGroup aVersionedProcessGroup() {
+ VersionedProcessGroup versionedProcessGroup = new VersionedProcessGroup();
+ VersionedProcessor versionedProcessor = new VersionedProcessor();
+ VersionedControllerService versionedControllerService = new VersionedControllerService();
+
+ versionedControllerService.setProperties(controllerServiceProperties);
+ versionedProcessor.setProperties(processorProperties);
+
+ versionedProcessGroup.setProcessors(Set.of(versionedProcessor));
+ versionedProcessGroup.setControllerServices(Set.of(versionedControllerService));
+
+ return versionedProcessGroup;
+ }
+}
\ No newline at end of file
diff --git a/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptorTest.java b/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptorTest.java
index 0b9a4c389f..9fb3c5c75e 100644
--- a/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptorTest.java
+++ b/minifi/minifi-commons/minifi-commons-framework/src/test/java/org/apache/nifi/minifi/commons/service/StandardFlowPropertyEncryptorTest.java
@@ -120,9 +120,9 @@ public class StandardFlowPropertyEncryptorTest {
public void shouldEncryptParameterContextsSensitiveVariables() {
VersionedDataflow testFlow = flowWithParameterContexts();
- VersionedDataflow encryptedFlow = testEncryptor.encryptSensitiveProperties(testFlow);
+ testEncryptor.encryptSensitiveProperties(testFlow);
- encryptedFlow.getParameterContexts().stream()
+ testFlow.getParameterContexts().stream()
.flatMap(context -> context.getParameters().stream())
.forEach(parameter -> {
if (parameter.isSensitive()) {
@@ -137,10 +137,10 @@ public class StandardFlowPropertyEncryptorTest {
public void shouldEncryptPropertiesUsingDescriptorsFromFlow() {
VersionedDataflow testFlow = flowWithPropertyDescriptors();
- VersionedDataflow encryptedFlow = testEncryptor.encryptSensitiveProperties(testFlow);
+ testEncryptor.encryptSensitiveProperties(testFlow);
verify(mockRunTimeManifest, never()).getBundles();
- assertSensitiveFlowComponentPropertiesAreEncoded(encryptedFlow);
+ assertSensitiveFlowComponentPropertiesAreEncoded(testFlow);
}
@Test
@@ -148,9 +148,9 @@ public class StandardFlowPropertyEncryptorTest {
VersionedDataflow testFlow = flowWithoutPropertyDescriptors();
when(mockRunTimeManifest.getBundles()).thenReturn(runTimeManifestBundles());
- VersionedDataflow encryptedFlow = testEncryptor.encryptSensitiveProperties(testFlow);
+ testEncryptor.encryptSensitiveProperties(testFlow);
- assertSensitiveFlowComponentPropertiesAreEncoded(encryptedFlow);
+ assertSensitiveFlowComponentPropertiesAreEncoded(testFlow);
}
private VersionedDataflow flowWithParameterContexts() {
diff --git a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java
index 97a5e31450..f2fa27b891 100644
--- a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java
+++ b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java
@@ -124,9 +124,12 @@ import org.apache.nifi.minifi.c2.command.UpdatePropertiesPropertyProvider;
import org.apache.nifi.minifi.c2.command.syncresource.DefaultSyncResourceStrategy;
import org.apache.nifi.minifi.c2.command.syncresource.FileResourceRepository;
import org.apache.nifi.minifi.c2.command.syncresource.ResourceRepository;
+import org.apache.nifi.minifi.c2.command.syncresource.SyncResourcePropertyProvider;
import org.apache.nifi.minifi.commons.api.MiNiFiProperties;
+import org.apache.nifi.minifi.commons.service.FlowPropertyAssetReferenceResolver;
import org.apache.nifi.minifi.commons.service.FlowPropertyEncryptor;
import org.apache.nifi.minifi.commons.service.StandardFlowEnrichService;
+import org.apache.nifi.minifi.commons.service.StandardFlowPropertyAssetReferenceResolverService;
import org.apache.nifi.minifi.commons.service.StandardFlowPropertyEncryptor;
import org.apache.nifi.minifi.commons.service.StandardFlowSerDeService;
import org.apache.nifi.nar.ExtensionManagerHolder;
@@ -248,14 +251,20 @@ public class C2NifiClientService {
UpdatePropertiesPropertyProvider updatePropertiesPropertyProvider = new UpdatePropertiesPropertyProvider(bootstrapConfigFileLocation);
PropertiesPersister propertiesPersister = new PropertiesPersister(updatePropertiesPropertyProvider, bootstrapConfigFileLocation);
FlowStateStrategy defaultFlowStateStrategy = new DefaultFlowStateStrategy(flowController);
+ FlowPropertyAssetReferenceResolver flowPropertyAssetReferenceResolver = new StandardFlowPropertyAssetReferenceResolverService(resourceRepository::getAbsolutePath);
FlowPropertyEncryptor flowPropertyEncryptor = new StandardFlowPropertyEncryptor(
new PropertyEncryptorBuilder(niFiProperties.getProperty(SENSITIVE_PROPS_KEY))
.setAlgorithm(niFiProperties.getProperty(SENSITIVE_PROPS_ALGORITHM)).build(),
runtimeManifestService.getManifest());
- UpdateConfigurationStrategy updateConfigurationStrategy = new DefaultUpdateConfigurationStrategy(flowController, flowService,
- new StandardFlowEnrichService(niFiProperties), flowPropertyEncryptor,
- StandardFlowSerDeService.defaultInstance(), niFiProperties.getProperty(FLOW_CONFIGURATION_FILE));
+ UpdateConfigurationStrategy updateConfigurationStrategy = new DefaultUpdateConfigurationStrategy(
+ flowController,
+ flowService,
+ flowPropertyAssetReferenceResolver,
+ new StandardFlowEnrichService(niFiProperties),
+ flowPropertyEncryptor,
+ StandardFlowSerDeService.defaultInstance(),
+ niFiProperties.getProperty(FLOW_CONFIGURATION_FILE));
Supplier<RuntimeInfoWrapper> runtimeInfoWrapperSupplier = () -> generateRuntimeInfo(
parseInt(niFiProperties.getProperty(C2_FLOW_INFO_PROCESSOR_BULLETIN_LIMIT.getKey(), C2_FLOW_INFO_PROCESSOR_BULLETIN_LIMIT.getDefaultValue())),
parseBoolean(niFiProperties.getProperty(C2_FLOW_INFO_PROCESSOR_STATUS_ENABLED.getKey(), C2_FLOW_INFO_PROCESSOR_STATUS_ENABLED.getDefaultValue())));
@@ -268,7 +277,7 @@ public class C2NifiClientService {
UpdateAssetOperationHandler.create(client, emptyOperandPropertiesProvider,
updateAssetCommandHelper::assetUpdatePrecondition, updateAssetCommandHelper::assetPersistFunction),
new UpdatePropertiesOperationHandler(updatePropertiesPropertyProvider, propertiesPersister::persistProperties),
- SyncResourceOperationHandler.create(client, emptyOperandPropertiesProvider, new DefaultSyncResourceStrategy(resourceRepository), c2Serializer),
+ SyncResourceOperationHandler.create(client, new SyncResourcePropertyProvider(), new DefaultSyncResourceStrategy(resourceRepository), c2Serializer),
new StartFlowOperationHandler(defaultFlowStateStrategy), new StopFlowOperationHandler(defaultFlowStateStrategy)
));
}
diff --git a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategy.java b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategy.java
index 00ceb59bae..37fdc79fcc 100644
--- a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategy.java
+++ b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategy.java
@@ -52,6 +52,7 @@ import org.apache.nifi.flow.VersionedProcessGroup;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.minifi.commons.service.FlowEnrichService;
+import org.apache.nifi.minifi.commons.service.FlowPropertyAssetReferenceResolver;
import org.apache.nifi.minifi.commons.service.FlowPropertyEncryptor;
import org.apache.nifi.minifi.commons.service.FlowSerDeService;
import org.apache.nifi.minifi.validator.ValidationException;
@@ -68,6 +69,7 @@ public class DefaultUpdateConfigurationStrategy implements UpdateConfigurationSt
private final FlowController flowController;
private final FlowService flowService;
+ private final FlowPropertyAssetReferenceResolver flowPropertyAssetReferenceResolver;
private final FlowEnrichService flowEnrichService;
private final FlowPropertyEncryptor flowPropertyEncryptor;
private final FlowSerDeService flowSerDeService;
@@ -76,10 +78,17 @@ public class DefaultUpdateConfigurationStrategy implements UpdateConfigurationSt
private final Path rawFlowConfigurationFile;
private final Path backupRawFlowConfigurationFile;
- public DefaultUpdateConfigurationStrategy(FlowController flowController, FlowService flowService, FlowEnrichService flowEnrichService,
- FlowPropertyEncryptor flowPropertyEncryptor, FlowSerDeService flowSerDeService, String flowConfigurationFile) {
+ public DefaultUpdateConfigurationStrategy(
+ FlowController flowController,
+ FlowService flowService,
+ FlowPropertyAssetReferenceResolver flowPropertyAssetReferenceResolver,
+ FlowEnrichService flowEnrichService,
+ FlowPropertyEncryptor flowPropertyEncryptor,
+ FlowSerDeService flowSerDeService,
+ String flowConfigurationFile) {
this.flowController = flowController;
this.flowService = flowService;
+ this.flowPropertyAssetReferenceResolver = flowPropertyAssetReferenceResolver;
this.flowEnrichService = flowEnrichService;
this.flowPropertyEncryptor = flowPropertyEncryptor;
this.flowSerDeService = flowSerDeService;
@@ -102,12 +111,13 @@ public class DefaultUpdateConfigurationStrategy implements UpdateConfigurationSt
.stream()
.map(Connection::getIdentifier)
.collect(Collectors.toSet());
- VersionedDataflow rawDataFlow = flowSerDeService.deserialize(rawFlow);
+ VersionedDataflow dataFlow = flowSerDeService.deserialize(rawFlow);
- VersionedDataflow propertyEncryptedRawDataFlow = flowPropertyEncryptor.encryptSensitiveProperties(rawDataFlow);
- byte[] serializedPropertyEncryptedRawDataFlow = flowSerDeService.serialize(propertyEncryptedRawDataFlow);
- VersionedDataflow enrichedFlowCandidate = flowEnrichService.enrichFlow(propertyEncryptedRawDataFlow);
- byte[] serializedEnrichedFlowCandidate = flowSerDeService.serialize(enrichedFlowCandidate);
+ flowPropertyAssetReferenceResolver.resolveAssetReferenceProperties(dataFlow);
+ flowPropertyEncryptor.encryptSensitiveProperties(dataFlow);
+ byte[] serializedPropertyEncryptedRawDataFlow = flowSerDeService.serialize(dataFlow);
+ flowEnrichService.enrichFlow(dataFlow);
+ byte[] serializedEnrichedFlowCandidate = flowSerDeService.serialize(dataFlow);
backup(flowConfigurationFile, backupFlowConfigurationFile);
backup(rawFlowConfigurationFile, backupRawFlowConfigurationFile);
@@ -115,7 +125,7 @@ public class DefaultUpdateConfigurationStrategy implements UpdateConfigurationSt
persist(serializedPropertyEncryptedRawDataFlow, rawFlowConfigurationFile, false);
persist(serializedEnrichedFlowCandidate, flowConfigurationFile, true);
- reloadFlow(findAllProposedConnectionIds(enrichedFlowCandidate.getRootGroup()));
+ reloadFlow(findAllProposedConnectionIds(dataFlow.getRootGroup()));
} catch (IllegalStateException e) {
LOGGER.error("Configuration update failed. Reverting and reloading previous flow", e);
diff --git a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepository.java b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepository.java
index 6ed8b2f3e4..a3a7791ec8 100644
--- a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepository.java
+++ b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepository.java
@@ -161,6 +161,16 @@ public class FileResourceRepository implements ResourceRepository {
return Optional.of(resourceItem);
}
+ @Override
+ public Optional<Path> getAbsolutePath(String resourceId) {
+ return resourceRepositoryDescriptor.resourceItems.stream()
+ .filter(resourceItem -> resourceItem.getResourceId().equals(resourceId))
+ .map(this::resourcePath)
+ .map(Path::toAbsolutePath)
+ .findFirst();
+ }
+
+
private void initialize() {
try {
createDirectories(assetRepositoryDirectory);
diff --git a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/ResourceRepository.java b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/ResourceRepository.java
index db431d361d..4a73b36fb3 100644
--- a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/ResourceRepository.java
+++ b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/ResourceRepository.java
@@ -38,4 +38,6 @@ public interface ResourceRepository {
Optional<ResourceItem> addResourceItem(ResourceItem resourceItem, Path source);
Optional<ResourceItem> deleteResourceItem(ResourceItem resourceItem);
+ Optional<Path> getAbsolutePath(String resourceId);
+
}
diff --git a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/SyncResourcePropertyProvider.java
similarity index 55%
copy from minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java
copy to minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/SyncResourcePropertyProvider.java
index 5b3620803f..87a5fe29be 100644
--- a/minifi/minifi-commons/minifi-commons-api/src/main/java/org/apache/nifi/minifi/commons/service/FlowEnrichService.java
+++ b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/SyncResourcePropertyProvider.java
@@ -15,21 +15,18 @@
* limitations under the License.
*/
-package org.apache.nifi.minifi.commons.service;
+package org.apache.nifi.minifi.c2.command.syncresource;
-import org.apache.nifi.controller.flow.VersionedDataflow;
+import org.apache.nifi.c2.client.service.operation.OperandPropertiesProvider;
+import java.util.Map;
-/**
- * Defines interface methods used to implement a FlowEnrichService.
- * The purpose of a flow enrich service is to enrich a VersionedDataFlow with various additional components specific to the MiNiFi instance
- */
-public interface FlowEnrichService {
+public class SyncResourcePropertyProvider implements OperandPropertiesProvider {
+
+ private static final String RESOLVE_ASSET_REFERENCES = "resolveAssetReferences";
+ private static final Map<String, Object> PROPERTIES = Map.of(RESOLVE_ASSET_REFERENCES, Boolean.TRUE);
- /**
- * Responsible for enriching a VersionedDataflow instance
- *
- * @param versionedDataflow a VersionedDataflow instance
- * @return VersionedDataflow the enriched flow instance
- */
- VersionedDataflow enrichFlow(VersionedDataflow versionedDataflow);
+ @Override
+ public Map<String, Object> getProperties() {
+ return PROPERTIES;
+ }
}
diff --git a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategyTest.java b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategyTest.java
index 6c2226f803..4addb65e7e 100644
--- a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategyTest.java
+++ b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/DefaultUpdateConfigurationStrategyTest.java
@@ -50,6 +50,7 @@ import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.controller.flow.VersionedDataflow;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.minifi.commons.service.FlowEnrichService;
+import org.apache.nifi.minifi.commons.service.FlowPropertyAssetReferenceResolver;
import org.apache.nifi.minifi.commons.service.FlowPropertyEncryptor;
import org.apache.nifi.minifi.commons.service.FlowSerDeService;
import org.apache.nifi.services.FlowService;
@@ -67,13 +68,10 @@ public class DefaultUpdateConfigurationStrategyTest {
private static final byte[] ORIGINAL_RAW_FLOW_CONFIG_CONTENT = "original_raw_content".getBytes(UTF_8);
private static final byte[] ORIGINAL_ENRICHED_FLOW_CONFIG_CONTENT = "original_enriched_content".getBytes(UTF_8);
-
private static final byte[] NEW_RAW_FLOW_CONFIG_CONTENT = "new_raw_content".getBytes(UTF_8);
- private static final VersionedDataflow NEW_RAW_FLOW_CONFIG = new VersionedDataflow();
+ private static final VersionedDataflow DATA_FLOW = new VersionedDataflow();
private static final byte[] NEW_ENCRYPTED_FLOW_CONFIG_CONTENT = "original_encrypted_content".getBytes(UTF_8);
- private static final VersionedDataflow NEW_ENCRYPTED_FLOW_CONFIG = new VersionedDataflow();
private static final byte[] NEW_ENRICHED_FLOW_CONFIG_CONTENT = "new_enriched_content".getBytes(UTF_8);
- private static final VersionedDataflow NEW_ENRICHED_FLOW_CONFIG = new VersionedDataflow();
@TempDir
private File tempDir;
@@ -87,6 +85,8 @@ public class DefaultUpdateConfigurationStrategyTest {
@Mock
private FlowPropertyEncryptor mockFlowPropertyEncryptor;
@Mock
+ private FlowPropertyAssetReferenceResolver mockFlowPropertyAssetReferenceResolver;
+ @Mock
private FlowSerDeService mockFlowSerDeService;
@Mock
private FlowManager mockFlowManager;
@@ -108,8 +108,14 @@ public class DefaultUpdateConfigurationStrategyTest {
rawFlowConfigurationFile = flowConfigurationFile.getParent().resolve(flowConfigurationFileBaseName + RAW_EXTENSION);
backupRawFlowConfigurationFile = flowConfigurationFile.getParent().resolve(flowConfigurationFileBaseName + BACKUP_EXTENSION + RAW_EXTENSION);
- testUpdateConfigurationStrategy = new DefaultUpdateConfigurationStrategy(mockFlowController, mockFlowService, mockFlowEnrichService,
- mockFlowPropertyEncryptor, mockFlowSerDeService, flowConfigurationFile.toString());
+ testUpdateConfigurationStrategy = new DefaultUpdateConfigurationStrategy(
+ mockFlowController,
+ mockFlowService,
+ mockFlowPropertyAssetReferenceResolver,
+ mockFlowEnrichService,
+ mockFlowPropertyEncryptor,
+ mockFlowSerDeService,
+ flowConfigurationFile.toString());
writeGzipFile(flowConfigurationFile, ORIGINAL_ENRICHED_FLOW_CONFIG_CONTENT);
writePlainTextFile(rawFlowConfigurationFile, ORIGINAL_RAW_FLOW_CONFIG_CONTENT);
@@ -118,11 +124,8 @@ public class DefaultUpdateConfigurationStrategyTest {
@Test
public void testFlowIsUpdatedAndBackupsAreClearedUp() throws IOException {
// given
- when(mockFlowSerDeService.deserialize(NEW_RAW_FLOW_CONFIG_CONTENT)).thenReturn(NEW_RAW_FLOW_CONFIG);
- when(mockFlowPropertyEncryptor.encryptSensitiveProperties(NEW_RAW_FLOW_CONFIG)).thenReturn(NEW_ENCRYPTED_FLOW_CONFIG);
- when(mockFlowSerDeService.serialize(NEW_ENCRYPTED_FLOW_CONFIG)).thenReturn(NEW_ENCRYPTED_FLOW_CONFIG_CONTENT);
- when(mockFlowEnrichService.enrichFlow(NEW_ENCRYPTED_FLOW_CONFIG)).thenReturn(NEW_ENRICHED_FLOW_CONFIG);
- when(mockFlowSerDeService.serialize(NEW_ENRICHED_FLOW_CONFIG)).thenReturn(NEW_ENRICHED_FLOW_CONFIG_CONTENT);
+ when(mockFlowSerDeService.deserialize(NEW_RAW_FLOW_CONFIG_CONTENT)).thenReturn(DATA_FLOW);
+ when(mockFlowSerDeService.serialize(DATA_FLOW)).thenReturn(NEW_ENCRYPTED_FLOW_CONFIG_CONTENT, NEW_ENRICHED_FLOW_CONFIG_CONTENT);
when(mockFlowController.getFlowManager()).thenReturn(mockFlowManager);
when(mockFlowManager.getRootGroup()).thenReturn(mockProcessGroup);
@@ -137,6 +140,9 @@ public class DefaultUpdateConfigurationStrategyTest {
assertFalse(exists(backupFlowConfigurationFile));
assertFalse(exists(backupRawFlowConfigurationFile));
verify(mockFlowService, times(1)).load(null);
+ verify(mockFlowPropertyAssetReferenceResolver, times(1)).resolveAssetReferenceProperties(DATA_FLOW);
+ verify(mockFlowEnrichService, times(1)).enrichFlow(DATA_FLOW);
+ verify(mockFlowPropertyEncryptor, times(1)).encryptSensitiveProperties(DATA_FLOW);
verify(mockFlowController, times(1)).onFlowInitialized(true);
verify(mockProcessGroup, times(1)).startProcessing();
}
@@ -144,11 +150,8 @@ public class DefaultUpdateConfigurationStrategyTest {
@Test
public void testFlowIsRevertedInCaseOfAnyErrorAndBackupsAreClearedUp() throws IOException {
// given
- when(mockFlowSerDeService.deserialize(NEW_RAW_FLOW_CONFIG_CONTENT)).thenReturn(NEW_RAW_FLOW_CONFIG);
- when(mockFlowPropertyEncryptor.encryptSensitiveProperties(NEW_RAW_FLOW_CONFIG)).thenReturn(NEW_ENCRYPTED_FLOW_CONFIG);
- when(mockFlowSerDeService.serialize(NEW_ENCRYPTED_FLOW_CONFIG)).thenReturn(NEW_ENCRYPTED_FLOW_CONFIG_CONTENT);
- when(mockFlowEnrichService.enrichFlow(NEW_ENCRYPTED_FLOW_CONFIG)).thenReturn(NEW_ENRICHED_FLOW_CONFIG);
- when(mockFlowSerDeService.serialize(NEW_ENRICHED_FLOW_CONFIG)).thenReturn(NEW_ENRICHED_FLOW_CONFIG_CONTENT);
+ when(mockFlowSerDeService.deserialize(NEW_RAW_FLOW_CONFIG_CONTENT)).thenReturn(DATA_FLOW);
+ when(mockFlowSerDeService.serialize(DATA_FLOW)).thenReturn(NEW_ENCRYPTED_FLOW_CONFIG_CONTENT, NEW_ENRICHED_FLOW_CONFIG_CONTENT);
when(mockFlowController.getFlowManager()).thenReturn(mockFlowManager);
when(mockFlowManager.getRootGroup()).thenReturn(mockProcessGroup);
doThrow(new IOException()).when(mockFlowService).load(null);
@@ -165,6 +168,9 @@ public class DefaultUpdateConfigurationStrategyTest {
assertFalse(exists(backupFlowConfigurationFile));
assertFalse(exists(backupRawFlowConfigurationFile));
verify(mockFlowService, times(1)).load(null);
+ verify(mockFlowEnrichService, times(1)).enrichFlow(DATA_FLOW);
+ verify(mockFlowPropertyEncryptor, times(1)).encryptSensitiveProperties(DATA_FLOW);
+ verify(mockFlowPropertyAssetReferenceResolver, times(1)).resolveAssetReferenceProperties(DATA_FLOW);
verify(mockFlowController, times(0)).onFlowInitialized(true);
verify(mockProcessGroup, times(0)).startProcessing();
}
diff --git a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepositoryTest.java b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepositoryTest.java
index ed13fb1648..47600be662 100644
--- a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepositoryTest.java
+++ b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepositoryTest.java
@@ -378,6 +378,28 @@ public class FileResourceRepositoryTest {
assertTrue(result.isEmpty());
}
+ @Test
+ public void testGetRelativePathReturnsEmptyInCaseOfResourceNotAvailable() throws IOException {
+ FileResourceRepository testRepository = createTestRepository();
+
+ Optional<Path> relativePath = testRepository.getAbsolutePath("non_existing_resource_id");
+
+ assertTrue(relativePath.isEmpty());
+ }
+
+ @Test
+ public void testGetRelativePath() throws IOException {
+ FileResourceRepository testRepository = createTestRepository();
+ ResourceItem resourceItem = resourceItem("resource1", "subfolder", ASSET);
+ Path resourceItemPath = createResourceBinary(resourceItem, RESOURCE_BINARY_CONTENT);
+ testRepository.addResourceItem(resourceItem);
+
+ Optional<Path> relativePath = testRepository.getAbsolutePath("resource1");
+
+ assertTrue(relativePath.isPresent());
+ assertEquals(resourceItemPath.toString(), relativePath.get().toString());
+ }
+
private ResourceRepositoryDescriptor loadRepository() throws IOException {
return c2Serializer.deserialize(readString(repositoryFile), ResourceRepositoryDescriptor.class).orElse(null);
}