This is an automated email from the ASF dual-hosted git repository.
briansolo1985 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 5c3e22aa44 NIFI-13614: populate failure cause in operation state
5c3e22aa44 is described below
commit 5c3e22aa44fd2f2883f4661d45449ce6a4bc70cd
Author: Kalman Jantner <[email protected]>
AuthorDate: Wed Aug 21 14:59:30 2024 +0200
NIFI-13614: populate failure cause in operation state
This closes #9136.
Signed-off-by: Ferenc Kis <[email protected]>
---
.../org/apache/nifi/c2/client/api/C2Client.java | 4 +-
.../apache/nifi/c2/client/http/C2HttpClient.java | 2 +-
.../nifi/c2/client/http/url/C2UrlProvider.java | 4 +-
.../c2/client/http/url/LegacyC2UrlProvider.java | 13 +++--
.../client/http/url/ProxyAwareC2UrlProvider.java | 5 +-
.../client/http/url/LegacyC2UrlProviderTest.java | 17 ++++--
.../client/http/url/ProxyAwareC2ProviderTest.java | 31 ++++++++---
.../service/operation/C2OperationHandler.java | 36 +++++++++++-
.../DescribeManifestOperationHandler.java | 26 ++++++---
.../operation/SyncResourceOperationHandler.java | 23 ++++----
.../service/operation/SyncResourceStrategy.java | 2 +-
.../operation/TransferDebugOperationHandler.java | 10 ++--
.../operation/UpdateAssetOperationHandler.java | 10 ++--
.../UpdateConfigurationOperationHandler.java | 16 ++++--
.../operation/UpdateConfigurationStrategy.java | 4 +-
.../nifi/minifi/validator/ValidationException.java | 56 +++++++++++++++++++
.../TransferDebugOperationHandlerTest.java | 5 +-
.../operation/UpdateAssetOperationHandlerTest.java | 4 +-
.../UpdateConfigurationOperationHandlerTest.java | 19 ++++---
.../nifi/c2/protocol/api/C2OperationState.java | 16 +++++-
.../apache/nifi/c2/protocol/api/FailureCause.java | 65 ++++++++++++++++++++++
.../DefaultUpdateConfigurationStrategy.java | 17 +++---
.../syncresource/DefaultSyncResourceStrategy.java | 8 +--
.../DefaultUpdateConfigurationStrategyTest.java | 30 +++++-----
.../DefaultSyncResourceStrategyTest.java | 2 +-
25 files changed, 319 insertions(+), 106 deletions(-)
diff --git
a/c2/c2-client-bundle/c2-client-api/src/main/java/org/apache/nifi/c2/client/api/C2Client.java
b/c2/c2-client-bundle/c2-client-api/src/main/java/org/apache/nifi/c2/client/api/C2Client.java
index ad46e2351c..425010355a 100644
---
a/c2/c2-client-bundle/c2-client-api/src/main/java/org/apache/nifi/c2/client/api/C2Client.java
+++
b/c2/c2-client-bundle/c2-client-api/src/main/java/org/apache/nifi/c2/client/api/C2Client.java
@@ -85,7 +85,7 @@ public interface C2Client {
*
* @param absoluteUrl absolute url sent by C2 server
* @param relativeUrl relative url sent by C2 server
- * @return an optional with content of finalised callback url
+ * @return finalised callback url
*/
- Optional<String> getCallbackUrl(String absoluteUrl, String relativeUrl);
+ String getCallbackUrl(String absoluteUrl, String relativeUrl);
}
diff --git
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java
index beff8d171a..1c9ecc163a 100644
---
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java
+++
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java
@@ -146,7 +146,7 @@ public class C2HttpClient implements C2Client {
}
@Override
- public Optional<String> getCallbackUrl(String absoluteUrl, String
relativeUrl) {
+ public String getCallbackUrl(String absoluteUrl, String relativeUrl) {
return c2UrlProvider.getCallbackUrl(absoluteUrl, relativeUrl);
}
diff --git
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/C2UrlProvider.java
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/C2UrlProvider.java
index 51ea7a30c8..15447f8b96 100644
---
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/C2UrlProvider.java
+++
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/C2UrlProvider.java
@@ -17,8 +17,6 @@
package org.apache.nifi.c2.client.http.url;
-import java.util.Optional;
-
public interface C2UrlProvider {
/**
@@ -42,5 +40,5 @@ public interface C2UrlProvider {
* @param relativeUrl relative url sent by the C2 server
* @return the url of the C2 server to send requests to
*/
- Optional<String> getCallbackUrl(String absoluteUrl, String relativeUrl);
+ String getCallbackUrl(String absoluteUrl, String relativeUrl);
}
diff --git
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProvider.java
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProvider.java
index fe9fad24ef..9040754486 100644
---
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProvider.java
+++
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProvider.java
@@ -45,11 +45,12 @@ public class LegacyC2UrlProvider implements C2UrlProvider {
}
@Override
- public Optional<String> getCallbackUrl(String absoluteUrl, String
relativeUrl) {
- Optional<String> url =
Optional.ofNullable(absoluteUrl).filter(StringUtils::isNotBlank);
- if (!url.isPresent()) {
- LOG.error("Provided absolute url was empty or null. Relative urls
are not supported with this configuration");
- }
- return url;
+ public String getCallbackUrl(String absoluteUrl, String relativeUrl) {
+ return Optional.ofNullable(absoluteUrl)
+ .filter(StringUtils::isNotBlank)
+ .orElseThrow( () -> {
+ LOG.error("Provided absolute url was empty or null.
Relative urls are not supported with this configuration");
+ throw new IllegalArgumentException("Provided absolute
url was empty or null. Relative C2 urls are not supported with this
configuration");
+ });
}
}
diff --git
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2UrlProvider.java
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2UrlProvider.java
index aeb8708101..fcb440081e 100644
---
a/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2UrlProvider.java
+++
b/c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2UrlProvider.java
@@ -59,11 +59,12 @@ public class ProxyAwareC2UrlProvider implements
C2UrlProvider {
}
@Override
- public Optional<String> getCallbackUrl(String absoluteUrl, String
relativeUrl) {
+ public String getCallbackUrl(String absoluteUrl, String relativeUrl) {
return Optional.ofNullable(relativeUrl)
.map(this::toAbsoluteUrl)
.filter(Optional::isPresent)
- .orElseGet(() ->
Optional.ofNullable(absoluteUrl).filter(StringUtils::isNotBlank));
+ .orElseGet(() ->
Optional.ofNullable(absoluteUrl).filter(StringUtils::isNotBlank))
+ .orElseThrow(() -> new IllegalArgumentException("Unable to return
non empty c2 url."));
}
private Optional<String> toAbsoluteUrl(String path) {
diff --git
a/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProviderTest.java
b/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProviderTest.java
index 05f88b65e1..bf19138457 100644
---
a/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProviderTest.java
+++
b/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/LegacyC2UrlProviderTest.java
@@ -18,6 +18,7 @@
package org.apache.nifi.c2.client.http.url;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import java.util.Optional;
import java.util.stream.Stream;
@@ -33,6 +34,9 @@ public class LegacyC2UrlProviderTest {
private static final String C2_HEARTBEAT_URL =
"https://host:8080/c2/api/heartbeat";;
private static final String C2_ACKNOWLEDGE_URL =
"https://host:8080/c2/api/acknowledge";;
+ private static final String ABSOLUTE_URL = "http://c2/api/callback";;
+ private static final String RELATIVE_URL = "any_url";
+ private static final String EXPECTED_URL = "http://c2/api/callback";;
@Test
public void
testProviderIsCreatedAndReturnsProperHeartbeatAndAcknowledgeUrls() {
@@ -44,9 +48,15 @@ public class LegacyC2UrlProviderTest {
@MethodSource("testCallbackUrlProvidedArguments")
@ParameterizedTest(name = "{index} => absoluteUrl={0}, relativeUrl={1},
expectedCallbackUrl={2}")
- public void testCallbackUrlProvidedFor(String absoluteUrl, String
relativeUrl, Optional<String> expectedCallbackUrl) {
+ public void testCallbackUrlProvidedForInvalidInputs(String absoluteUrl,
String relativeUrl) {
LegacyC2UrlProvider testProvider = new
LegacyC2UrlProvider(C2_HEARTBEAT_URL, C2_ACKNOWLEDGE_URL);
- assertEquals(expectedCallbackUrl,
testProvider.getCallbackUrl(absoluteUrl, relativeUrl));
+ assertThrows(IllegalArgumentException.class, () ->
testProvider.getCallbackUrl(absoluteUrl, relativeUrl));
+ }
+
+ @Test
+ public void testCallbackUrlProvidedFor() {
+ LegacyC2UrlProvider testProvider = new
LegacyC2UrlProvider(C2_HEARTBEAT_URL, C2_ACKNOWLEDGE_URL);
+ assertEquals(EXPECTED_URL, testProvider.getCallbackUrl(ABSOLUTE_URL,
RELATIVE_URL));
}
private static Stream<Arguments> testCallbackUrlProvidedArguments() {
@@ -54,8 +64,7 @@ public class LegacyC2UrlProviderTest {
Arguments.of(null, null, Optional.empty()),
Arguments.of(null, "any_url", Optional.empty()),
Arguments.of("", "", Optional.empty()),
- Arguments.of("", "any_url", Optional.empty()),
- Arguments.of("http://c2/api/callback";, "any_url",
Optional.of("http://c2/api/callback";))
+ Arguments.of("", "any_url", Optional.empty())
);
}
}
diff --git
a/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2ProviderTest.java
b/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2ProviderTest.java
index 99c1a82211..21dffc4328 100644
---
a/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2ProviderTest.java
+++
b/c2/c2-client-bundle/c2-client-http/src/test/java/org/apache/nifi/c2/client/http/url/ProxyAwareC2ProviderTest.java
@@ -18,6 +18,7 @@
package org.apache.nifi.c2.client.http.url;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
import java.util.Optional;
@@ -83,16 +84,36 @@ public class ProxyAwareC2ProviderTest {
@MethodSource("testCallbackUrlProvidedArguments")
@ParameterizedTest(name = "{index} => c2RestBase={0}, absoluteUrl={1},
relativeUrl={2}, expectedCallbackUrl={3}")
- public void testCallbackUrlProvidedFor(String c2RestBase, String
absoluteUrl, String relativeUrl, Optional<String> expectedCallbackUrl) {
+ public void testCallbackUrlProvidedForValidInputs(String c2RestBase,
String absoluteUrl, String relativeUrl, String expectedCallbackUrl) {
ProxyAwareC2UrlProvider testProvider = new
ProxyAwareC2UrlProvider(c2RestBase, "any_path", "any_path");
assertEquals(expectedCallbackUrl,
testProvider.getCallbackUrl(absoluteUrl, relativeUrl));
}
+ @MethodSource("testCallbackUrlProvidedInvalidArguments")
+ @ParameterizedTest(name = "{index} => c2RestBase={0}, absoluteUrl={1},
relativeUrl={2}, expectedCallbackUrl={3}")
+ public void testCallbackUrlProvidedForInvalidInputs(String c2RestBase,
String absoluteUrl, String relativeUrl) {
+ ProxyAwareC2UrlProvider testProvider = new
ProxyAwareC2UrlProvider(c2RestBase, "any_path", "any_path");
+ assertThrows(IllegalArgumentException.class, () ->
testProvider.getCallbackUrl(absoluteUrl, relativeUrl));
+ }
+
private static Stream<Arguments> testCallbackUrlProvidedArguments() {
String c2RestBaseNoTrailingSlash = "http://c2/api";;
String c2RestBaseWithTrailingSlash = "http://c2/api/";;
String path = "path/endpoint";
String absoluteUrl = "http://c2-other/api/path/endpoint";;
+ return Stream.of(
+ Arguments.of(c2RestBaseNoTrailingSlash, null, path,
c2RestBaseWithTrailingSlash + path),
+ Arguments.of(c2RestBaseNoTrailingSlash, "", "/" + path,
c2RestBaseWithTrailingSlash + path),
+ Arguments.of(c2RestBaseWithTrailingSlash, null, path,
c2RestBaseWithTrailingSlash + path),
+ Arguments.of(c2RestBaseWithTrailingSlash, "", "/" + path,
c2RestBaseWithTrailingSlash + path),
+ Arguments.of(c2RestBaseWithTrailingSlash, absoluteUrl, null,
absoluteUrl),
+ Arguments.of(c2RestBaseWithTrailingSlash, absoluteUrl, "",
absoluteUrl)
+ );
+ }
+
+ private static Stream<Arguments> testCallbackUrlProvidedInvalidArguments()
{
+ String c2RestBaseNoTrailingSlash = "http://c2/api";;
+ String c2RestBaseWithTrailingSlash = "http://c2/api/";;
return Stream.of(
Arguments.of(c2RestBaseNoTrailingSlash, null, null,
Optional.empty()),
Arguments.of(c2RestBaseNoTrailingSlash, "", null,
Optional.empty()),
@@ -101,13 +122,7 @@ public class ProxyAwareC2ProviderTest {
Arguments.of(c2RestBaseWithTrailingSlash, null, null,
Optional.empty()),
Arguments.of(c2RestBaseWithTrailingSlash, "", null,
Optional.empty()),
Arguments.of(c2RestBaseWithTrailingSlash, null, "",
Optional.empty()),
- Arguments.of(c2RestBaseWithTrailingSlash, "", "",
Optional.empty()),
- Arguments.of(c2RestBaseNoTrailingSlash, null, path,
Optional.of(c2RestBaseWithTrailingSlash + path)),
- Arguments.of(c2RestBaseNoTrailingSlash, "", "/" + path,
Optional.of(c2RestBaseWithTrailingSlash + path)),
- Arguments.of(c2RestBaseWithTrailingSlash, null, path,
Optional.of(c2RestBaseWithTrailingSlash + path)),
- Arguments.of(c2RestBaseWithTrailingSlash, "", "/" + path,
Optional.of(c2RestBaseWithTrailingSlash + path)),
- Arguments.of(c2RestBaseWithTrailingSlash, absoluteUrl, null,
Optional.of(absoluteUrl)),
- Arguments.of(c2RestBaseWithTrailingSlash, absoluteUrl, "",
Optional.of(absoluteUrl))
+ Arguments.of(c2RestBaseWithTrailingSlash, "", "", Optional.empty())
);
}
}
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/C2OperationHandler.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/C2OperationHandler.java
index 4ecd662734..4d504d048e 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/C2OperationHandler.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/C2OperationHandler.java
@@ -20,14 +20,18 @@ package org.apache.nifi.c2.client.service.operation;
import static java.util.Optional.ofNullable;
import com.fasterxml.jackson.core.type.TypeReference;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.apache.nifi.c2.protocol.api.C2Operation;
import org.apache.nifi.c2.protocol.api.C2OperationAck;
import org.apache.nifi.c2.protocol.api.C2OperationState;
+import org.apache.nifi.c2.protocol.api.FailureCause;
import org.apache.nifi.c2.protocol.api.OperandType;
import org.apache.nifi.c2.protocol.api.OperationType;
import org.apache.nifi.c2.serializer.C2Serializer;
+import org.apache.nifi.minifi.validator.ValidationException;
/**
* Handler interface for the different operation types
@@ -79,13 +83,38 @@ public interface C2OperationHandler {
* @param details additional status info to detail the state
* @return the created state
*/
- default C2OperationState operationState(C2OperationState.OperationState
operationState, String details) {
+ default C2OperationState operationState(C2OperationState.OperationState
operationState, String details, Exception e) {
C2OperationState state = new C2OperationState();
state.setState(operationState);
state.setDetails(details);
+
ofNullable(e).map(this::toFailureCause).ifPresent(state::setFailureCause);
return state;
}
+ private FailureCause toFailureCause(Exception exception) {
+ FailureCause failureCause = new FailureCause();
+ failureCause.setExceptionMessage(exception.getMessage());
+ List<String> causeList = new LinkedList<>();
+ populateCausedChain(ofNullable(exception.getCause()), causeList);
+ failureCause.setCausedByMessages(causeList);
+ if (exception instanceof ValidationException validationException) {
+
failureCause.setValidationResults(validationException.getValidationResults());
+ }
+ return failureCause;
+ }
+
+ private List<String> populateCausedChain(Optional<Throwable> cause,
List<String> causeList) {
+ cause.ifPresent(c -> {
+ causeList.add(c.getMessage());
+ populateCausedChain(cause.map(Throwable::getCause), causeList);
+ });
+ return causeList;
+ }
+
+ default C2OperationState operationState(C2OperationState.OperationState
operationState, String details) {
+ return operationState(operationState, details, null);
+ }
+
/**
* Commonly used logic for creating an C2OperationAck object
*
@@ -122,9 +151,10 @@ public interface C2OperationHandler {
* @param serializer the serializer used to converting to the target class
* @return the optional retrieved and converted argument value
*/
- default <T> Optional<T> getOperationArg(C2Operation operation, String
argument, TypeReference<T> type, C2Serializer serializer) {
+ default <T> T getOperationArg(C2Operation operation, String argument,
TypeReference<T> type, C2Serializer serializer) {
return ofNullable(operation.getArgs())
.map(args -> args.get(argument))
- .flatMap(arg -> serializer.convert(arg, type));
+ .flatMap(arg -> serializer.convert(arg, type))
+ .orElseThrow(() -> new IllegalArgumentException("Failed to parse
argument " + argument + " of operation " + operation));
}
}
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/DescribeManifestOperationHandler.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/DescribeManifestOperationHandler.java
index 41fc564613..a443226acb 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/DescribeManifestOperationHandler.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/DescribeManifestOperationHandler.java
@@ -20,6 +20,7 @@ package org.apache.nifi.c2.client.service.operation;
import static java.util.Optional.ofNullable;
import static org.apache.commons.lang3.StringUtils.EMPTY;
import static
org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static
org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
import static org.apache.nifi.c2.protocol.api.OperandType.MANIFEST;
import static org.apache.nifi.c2.protocol.api.OperationType.DESCRIBE;
@@ -33,12 +34,16 @@ import org.apache.nifi.c2.protocol.api.C2Operation;
import org.apache.nifi.c2.protocol.api.C2OperationAck;
import org.apache.nifi.c2.protocol.api.OperandType;
import org.apache.nifi.c2.protocol.api.OperationType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class DescribeManifestOperationHandler implements C2OperationHandler {
+ private static final String ERROR_MESSAGE = "Failed to execute manifest
describe operation.";
private final C2HeartbeatFactory heartbeatFactory;
private final Supplier<RuntimeInfoWrapper> runtimeInfoSupplier;
private final OperandPropertiesProvider operandPropertiesProvider;
+ private static final Logger LOGGER =
LoggerFactory.getLogger(DescribeManifestOperationHandler.class);
public DescribeManifestOperationHandler(C2HeartbeatFactory
heartbeatFactory, Supplier<RuntimeInfoWrapper> runtimeInfoSupplier,
OperandPropertiesProvider
operandPropertiesProvider) {
@@ -60,15 +65,20 @@ public class DescribeManifestOperationHandler implements
C2OperationHandler {
@Override
public C2OperationAck handle(C2Operation operation) {
String operationId =
ofNullable(operation.getIdentifier()).orElse(EMPTY);
+ C2OperationAck c2OperationAck;
+ try {
+ RuntimeInfoWrapper runtimeInfoWrapper = runtimeInfoSupplier.get();
+ C2Heartbeat heartbeat =
heartbeatFactory.create(runtimeInfoWrapper);
- RuntimeInfoWrapper runtimeInfoWrapper = runtimeInfoSupplier.get();
- C2Heartbeat heartbeat = heartbeatFactory.create(runtimeInfoWrapper);
-
- C2OperationAck c2OperationAck = operationAck(operationId,
operationState(FULLY_APPLIED, EMPTY));
- c2OperationAck.setAgentInfo(agentInfo(heartbeat, runtimeInfoWrapper));
- c2OperationAck.setDeviceInfo(heartbeat.getDeviceInfo());
- c2OperationAck.setFlowInfo(heartbeat.getFlowInfo());
- c2OperationAck.setResourceInfo(heartbeat.getResourceInfo());
+ c2OperationAck = operationAck(operationId,
operationState(FULLY_APPLIED, EMPTY));
+ c2OperationAck.setAgentInfo(agentInfo(heartbeat,
runtimeInfoWrapper));
+ c2OperationAck.setDeviceInfo(heartbeat.getDeviceInfo());
+ c2OperationAck.setFlowInfo(heartbeat.getFlowInfo());
+ c2OperationAck.setResourceInfo(heartbeat.getResourceInfo());
+ } catch (Exception e) {
+ LOGGER.error(ERROR_MESSAGE, e);
+ c2OperationAck = operationAck(operationId,
operationState(NOT_APPLIED, ERROR_MESSAGE, e));
+ }
return c2OperationAck;
}
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceOperationHandler.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceOperationHandler.java
index 839c287a25..01153be856 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceOperationHandler.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceOperationHandler.java
@@ -27,7 +27,6 @@ import static org.apache.nifi.c2.util.Preconditions.requires;
import com.fasterxml.jackson.core.type.TypeReference;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import org.apache.nifi.c2.client.api.C2Client;
import org.apache.nifi.c2.protocol.api.C2Operation;
import org.apache.nifi.c2.protocol.api.C2OperationAck;
@@ -89,20 +88,24 @@ public class SyncResourceOperationHandler implements
C2OperationHandler {
public C2OperationAck handle(C2Operation operation) {
String operationId =
ofNullable(operation.getIdentifier()).orElse(EMPTY);
- Optional<ResourcesGlobalHash> resourcesGlobalHash =
getOperationArg(operation, GLOBAL_HASH_FIELD, new TypeReference<>() {
- }, c2Serializer);
- if (resourcesGlobalHash.isEmpty()) {
+ ResourcesGlobalHash resourcesGlobalHash;
+ try {
+ resourcesGlobalHash = getOperationArg(operation,
GLOBAL_HASH_FIELD, new TypeReference<>() { }, c2Serializer);
+ } catch (Exception e) {
LOG.error("Resources global hash could not be constructed from C2
request");
- return operationAck(operationId, operationState(NOT_APPLIED,
"Resources global hash element was not found"));
+ return operationAck(operationId, operationState(NOT_APPLIED,
"Resources global hash element was not found", e));
}
- Optional<List<ResourceItem>> resourceItems =
getOperationArg(operation, RESOURCE_LIST_FIELD, new TypeReference<>() {
- }, c2Serializer);
- if (resourceItems.isEmpty()) {
+
+ List<ResourceItem> resourceItems;
+ try {
+ resourceItems = getOperationArg(operation, RESOURCE_LIST_FIELD, new
TypeReference<>() { }, c2Serializer);
+
+ } catch (Exception e) {
LOG.error("Resource item list could not be constructed from C2
request");
- return operationAck(operationId, operationState(NOT_APPLIED,
"Resource item list element was not found"));
+ return operationAck(operationId, operationState(NOT_APPLIED,
"Resource item list element was not found", e));
}
- OperationState operationState =
syncResourceStrategy.synchronizeResourceRepository(resourcesGlobalHash.get(),
resourceItems.get(), c2Client::retrieveResourceItem,
+ OperationState operationState =
syncResourceStrategy.synchronizeResourceRepository(resourcesGlobalHash,
resourceItems, c2Client::retrieveResourceItem,
relativeUrl -> c2Client.getCallbackUrl(null, relativeUrl));
C2OperationState resultState = operationState(
operationState,
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceStrategy.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceStrategy.java
index 213614204a..75255dc2f1 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceStrategy.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceStrategy.java
@@ -31,5 +31,5 @@ public interface SyncResourceStrategy {
OperationState synchronizeResourceRepository(ResourcesGlobalHash
resourcesGlobalHash, List<ResourceItem> c2ServerItems,
BiFunction<String,
Function<InputStream, Optional<Path>>, Optional<Path>> resourceDownloadFunction,
- Function<String,
Optional<String>> urlEnrichFunction);
+ Function<String, String>
urlEnrichFunction);
}
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandler.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandler.java
index 083464d578..99d7bbb1f5 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandler.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandler.java
@@ -113,10 +113,12 @@ public class TransferDebugOperationHandler implements
C2OperationHandler {
public C2OperationAck handle(C2Operation operation) {
String operationId =
ofNullable(operation.getIdentifier()).orElse(EMPTY);
- Optional<String> callbackUrl =
c2Client.getCallbackUrl(getOperationArg(operation, TARGET_ARG).orElse(EMPTY),
getOperationArg(operation, RELATIVE_TARGET_ARG).orElse(EMPTY));
- if (callbackUrl.isEmpty()) {
+ String callbackUrl;
+ try {
+ callbackUrl = c2Client.getCallbackUrl(getOperationArg(operation,
TARGET_ARG).orElse(EMPTY), getOperationArg(operation,
RELATIVE_TARGET_ARG).orElse(EMPTY));
+ } catch (Exception e) {
LOG.error("Callback URL could not be constructed from C2 request
and current configuration");
- return operationAck(operationId, operationState(NOT_APPLIED,
C2_CALLBACK_URL_NOT_FOUND));
+ return operationAck(operationId, operationState(NOT_APPLIED,
C2_CALLBACK_URL_NOT_FOUND, e));
}
List<Path> preparedFiles = null;
@@ -124,7 +126,7 @@ public class TransferDebugOperationHandler implements
C2OperationHandler {
try {
preparedFiles = prepareFiles(operationId, bundleFilePaths);
operationState = createDebugBundle(preparedFiles)
- .map(bundle -> c2Client.uploadBundle(callbackUrl.get(), bundle)
+ .map(bundle -> c2Client.uploadBundle(callbackUrl, bundle)
.map(errorMessage -> operationState(NOT_APPLIED,
errorMessage))
.orElseGet(() -> operationState(FULLY_APPLIED,
SUCCESSFUL_UPLOAD)))
.orElseGet(() -> operationState(NOT_APPLIED,
UNABLE_TO_CREATE_BUNDLE));
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandler.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandler.java
index 76847a58c1..fb70d69714 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandler.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandler.java
@@ -97,9 +97,11 @@ public class UpdateAssetOperationHandler implements
C2OperationHandler {
public C2OperationAck handle(C2Operation operation) {
String operationId =
ofNullable(operation.getIdentifier()).orElse(EMPTY);
- Optional<String> callbackUrl =
- c2Client.getCallbackUrl(getOperationArg(operation,
ASSET_URL_KEY).orElse(EMPTY), getOperationArg(operation,
ASSET_RELATIVE_URL_KEY).orElse(EMPTY));
- if (callbackUrl.isEmpty()) {
+ String callbackUrl;
+
+ try {
+ callbackUrl = c2Client.getCallbackUrl(getOperationArg(operation,
ASSET_URL_KEY).orElse(EMPTY), getOperationArg(operation,
ASSET_RELATIVE_URL_KEY).orElse(EMPTY));
+ } catch (Exception e) {
LOG.error("Callback URL could not be constructed from C2 request
and current configuration");
return operationAck(operationId, operationState(NOT_APPLIED,
C2_CALLBACK_URL_NOT_FOUND));
}
@@ -114,7 +116,7 @@ public class UpdateAssetOperationHandler implements
C2OperationHandler {
LOG.info("Initiating asset update from url {} with name {}, force
update is {}", callbackUrl, assetFileName, forceDownload);
C2OperationState operationState =
assetUpdatePrecondition.test(assetFileName.get(), forceDownload)
- ? c2Client.retrieveUpdateAssetContent(callbackUrl.get())
+ ? c2Client.retrieveUpdateAssetContent(callbackUrl)
.map(content -> assetPersistFunction.apply(assetFileName.get(),
content)
? operationState(FULLY_APPLIED, SUCCESSFULLY_UPDATE_ASSET)
: operationState(NOT_APPLIED, FAILED_TO_PERSIST_ASSET_TO_DISK))
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandler.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandler.java
index ecc31920a4..57e343fe9c 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandler.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandler.java
@@ -91,13 +91,15 @@ public class UpdateConfigurationOperationHandler implements
C2OperationHandler {
String operationId =
ofNullable(operation.getIdentifier()).orElse(EMPTY);
String absoluteFlowUrl = getOperationArg(operation,
FLOW_URL_KEY).orElse(getOperationArg(operation, LOCATION).orElse(EMPTY));
- Optional<String> callbackUrl = client.getCallbackUrl(absoluteFlowUrl,
getOperationArg(operation, FLOW_RELATIVE_URL_KEY).orElse(EMPTY));
- if (callbackUrl.isEmpty()) {
+ String callbackUrl;
+ try {
+ callbackUrl = client.getCallbackUrl(absoluteFlowUrl,
getOperationArg(operation, FLOW_RELATIVE_URL_KEY).orElse(EMPTY));
+ } catch (Exception e) {
logger.error("Callback URL could not be constructed from C2
request and current configuration");
return operationAck(operationId, operationState(NOT_APPLIED,
"Could not get callback url from operation and current configuration"));
}
- Optional<String> flowId = getFlowId(operation, callbackUrl.get());
+ Optional<String> flowId = getFlowId(operation, callbackUrl);
if (flowId.isEmpty()) {
logger.error("FlowId is missing, no update will be performed");
return operationAck(operationId, operationState(NOT_APPLIED,
"Could not get flowId from the operation"));
@@ -110,7 +112,7 @@ public class UpdateConfigurationOperationHandler implements
C2OperationHandler {
logger.info("Will perform flow update from {} for operation #{}.
Previous flow id was {}, replacing with new id {}",
callbackUrl, operationId,
ofNullable(flowIdHolder.getFlowId()).orElse("not set"), flowId.get());
- C2OperationState state = updateFlow(operationId, callbackUrl.get());
+ C2OperationState state = updateFlow(operationId, callbackUrl);
if (state.getState() == FULLY_APPLIED) {
flowIdHolder.setFlowId(flowId.get());
}
@@ -125,9 +127,11 @@ public class UpdateConfigurationOperationHandler
implements C2OperationHandler {
return operationState(NOT_APPLIED, "Update content retrieval
resulted in empty content");
}
- if (!updateConfigurationStrategy.update(updateContent.get())) {
+ try {
+ updateConfigurationStrategy.update(updateContent.get());
+ } catch (Exception e) {
logger.error("Update resulted in error for operation #{}.",
opIdentifier);
- return operationState(NOT_APPLIED, "Update resulted in error");
+ return operationState(NOT_APPLIED, "Update resulted in error:", e);
}
logger.debug("Update configuration applied for operation #{}.",
opIdentifier);
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationStrategy.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationStrategy.java
index e551cb853e..32c82dc61f 100644
---
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationStrategy.java
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationStrategy.java
@@ -27,7 +27,7 @@ public interface UpdateConfigurationStrategy {
* Updates the MiNiFi agent's flow with the flow passed as parameter
*
* @param flow the MiNiFi flow config JSON represented as a byte array
- * @return true if the flow update was true, false otherwise
+ * @throw exception if update failed.
*/
- boolean update(byte[] flow);
+ void update(byte[] flow);
}
diff --git
a/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/minifi/validator/ValidationException.java
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/minifi/validator/ValidationException.java
new file mode 100644
index 0000000000..ac542a793c
--- /dev/null
+++
b/c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/minifi/validator/ValidationException.java
@@ -0,0 +1,56 @@
+/*
+ * 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.validator;
+
+import java.util.List;
+import java.util.Objects;
+import org.apache.nifi.components.ValidationResult;
+
+public class ValidationException extends IllegalStateException {
+ private List<ValidationResult> validationResults;
+
+ public ValidationException(String message, List<ValidationResult> details)
{
+ super(message);
+ this.validationResults = details;
+ }
+
+ public List<ValidationResult> getValidationResults() {
+ return validationResults;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ ValidationException that = (ValidationException) o;
+ return Objects.equals(validationResults, that.validationResults);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(validationResults);
+ }
+
+ @Override
+ public String toString() {
+ return "ValidationException{" +
+ "validationResults=" + validationResults +
+ "message=" + this.getMessage() +
+ '}';
+ }
+}
diff --git
a/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandlerTest.java
b/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandlerTest.java
index 78a0fc4426..07d83329f3 100644
---
a/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandlerTest.java
+++
b/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/TransferDebugOperationHandlerTest.java
@@ -51,7 +51,6 @@ import java.nio.file.Paths;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.function.Predicate;
import java.util.stream.Stream;
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
@@ -139,7 +138,7 @@ public class TransferDebugOperationHandlerTest {
.collect(toList());
TransferDebugOperationHandler testHandler =
TransferDebugOperationHandler.create(c2Client, operandPropertiesProvider,
createBundleFiles, DEFAULT_CONTENT_FILTER);
C2Operation c2Operation = operation(C2_DEBUG_UPLOAD_ENDPOINT);
- when(c2Client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(C2_DEBUG_UPLOAD_ENDPOINT));
+ when(c2Client.getCallbackUrl(any(),
any())).thenReturn(C2_DEBUG_UPLOAD_ENDPOINT);
// when
C2OperationAck result = testHandler.handle(c2Operation);
@@ -200,7 +199,7 @@ public class TransferDebugOperationHandlerTest {
Predicate<String> testContentFilter = content ->
!content.contains(filterKeyword);
TransferDebugOperationHandler testHandler =
TransferDebugOperationHandler.create(c2Client, operandPropertiesProvider,
singletonList(bundleFile), testContentFilter);
C2Operation c2Operation = operation(C2_DEBUG_UPLOAD_ENDPOINT);
- when(c2Client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(C2_DEBUG_UPLOAD_ENDPOINT));
+ when(c2Client.getCallbackUrl(any(),
any())).thenReturn(C2_DEBUG_UPLOAD_ENDPOINT);
// when
C2OperationAck result = testHandler.handle(c2Operation);
diff --git
a/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandlerTest.java
b/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandlerTest.java
index 4a171dc8f1..f3754a0b42 100644
---
a/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandlerTest.java
+++
b/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateAssetOperationHandlerTest.java
@@ -94,7 +94,7 @@ public class UpdateAssetOperationHandlerTest {
@BeforeEach
public void setup() {
- lenient().when(c2Client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(ASSET_URL));
+ lenient().when(c2Client.getCallbackUrl(any(),
any())).thenReturn(ASSET_URL);
}
@ParameterizedTest(name = "c2Client={0} operandPropertiesProvider={1}
bundleFileList={2} contentFilter={3}")
@@ -115,7 +115,7 @@ public class UpdateAssetOperationHandlerTest {
public void testAssetUrlCanNotBeNull() {
// given
C2Operation operation = operation(null, ASSET_FILE_NAME,
FORCE_DOWNLOAD);
- when(c2Client.getCallbackUrl(any(),
any())).thenReturn(Optional.empty());
+ when(c2Client.getCallbackUrl(any(), any())).thenThrow(new
IllegalArgumentException());
// when
C2OperationAck result = testHandler.handle(operation);
diff --git
a/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandlerTest.java
b/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandlerTest.java
index d67d56cdc8..a358fbeb89 100644
---
a/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandlerTest.java
+++
b/c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandlerTest.java
@@ -22,6 +22,8 @@ import static
org.apache.nifi.c2.client.service.operation.UpdateConfigurationOpe
import static
org.apache.nifi.c2.client.service.operation.UpdateConfigurationOperationHandler.LOCATION;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -72,7 +74,7 @@ public class UpdateConfigurationOperationHandlerTest {
C2Operation operation = new C2Operation();
operation.setArgs(INCORRECT_LOCATION_MAP);
- when(client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(INCORRECT_LOCATION));
+ when(client.getCallbackUrl(any(),
any())).thenReturn(INCORRECT_LOCATION);
C2OperationAck response = handler.handle(operation);
@@ -81,10 +83,10 @@ public class UpdateConfigurationOperationHandlerTest {
@Test
void testHandleFlowIdInArg() {
- UpdateConfigurationStrategy successUpdate = flow -> true;
+ UpdateConfigurationStrategy successUpdate =
mock(UpdateConfigurationStrategy.class);
when(flowIdHolder.getFlowId()).thenReturn(FLOW_ID);
when(client.retrieveUpdateConfigurationContent(any())).thenReturn(Optional.of("content".getBytes()));
- when(client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(INCORRECT_LOCATION));
+ when(client.getCallbackUrl(any(),
any())).thenReturn(INCORRECT_LOCATION);
UpdateConfigurationOperationHandler handler = new
UpdateConfigurationOperationHandler(client, flowIdHolder, successUpdate,
operandPropertiesProvider);
C2Operation operation = new C2Operation();
operation.setIdentifier(OPERATION_ID);
@@ -103,7 +105,7 @@ public class UpdateConfigurationOperationHandlerTest {
@Test
void testHandleReturnsNoOperationWithNoContent() {
when(flowIdHolder.getFlowId()).thenReturn(FLOW_ID);
- when(client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(CORRECT_LOCATION));
+ when(client.getCallbackUrl(any(), any())).thenReturn(CORRECT_LOCATION);
UpdateConfigurationOperationHandler handler = new
UpdateConfigurationOperationHandler(client, flowIdHolder, null,
operandPropertiesProvider);
C2Operation operation = new C2Operation();
operation.setArgs(CORRECT_LOCATION_MAP);
@@ -116,10 +118,11 @@ public class UpdateConfigurationOperationHandlerTest {
@Test
void testHandleReturnsNotAppliedWithContentApplyIssues() {
- UpdateConfigurationStrategy failedToUpdate = flow -> false;
+ UpdateConfigurationStrategy failedToUpdate =
mock(UpdateConfigurationStrategy.class);
+ doThrow(new
IllegalStateException()).when(failedToUpdate).update(any());
when(flowIdHolder.getFlowId()).thenReturn("previous_flow_id");
when(client.retrieveUpdateConfigurationContent(any())).thenReturn(Optional.of("content".getBytes()));
- when(client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(CORRECT_LOCATION));
+ when(client.getCallbackUrl(any(), any())).thenReturn(CORRECT_LOCATION);
UpdateConfigurationOperationHandler handler = new
UpdateConfigurationOperationHandler(client, flowIdHolder, failedToUpdate,
operandPropertiesProvider);
C2Operation operation = new C2Operation();
operation.setIdentifier(OPERATION_ID);
@@ -133,9 +136,9 @@ public class UpdateConfigurationOperationHandlerTest {
@Test
void testHandleReturnsFullyApplied() {
- UpdateConfigurationStrategy successUpdate = flow -> true;
+ UpdateConfigurationStrategy successUpdate =
mock(UpdateConfigurationStrategy.class);
when(flowIdHolder.getFlowId()).thenReturn("previous_flow_id");
- when(client.getCallbackUrl(any(),
any())).thenReturn(Optional.of(CORRECT_LOCATION));
+ when(client.getCallbackUrl(any(), any())).thenReturn(CORRECT_LOCATION);
when(client.retrieveUpdateConfigurationContent(any())).thenReturn(Optional.of("content".getBytes()));
UpdateConfigurationOperationHandler handler = new
UpdateConfigurationOperationHandler(client, flowIdHolder, successUpdate,
operandPropertiesProvider);
C2Operation operation = new C2Operation();
diff --git
a/c2/c2-protocol/c2-protocol-api/src/main/java/org/apache/nifi/c2/protocol/api/C2OperationState.java
b/c2/c2-protocol/c2-protocol-api/src/main/java/org/apache/nifi/c2/protocol/api/C2OperationState.java
index 1e9c1e5bf2..7414b53c11 100644
---
a/c2/c2-protocol/c2-protocol-api/src/main/java/org/apache/nifi/c2/protocol/api/C2OperationState.java
+++
b/c2/c2-protocol/c2-protocol-api/src/main/java/org/apache/nifi/c2/protocol/api/C2OperationState.java
@@ -42,6 +42,17 @@ public class C2OperationState implements Serializable {
@Schema(description = "Additional details about the state")
private String details;
+ @Schema(description = "Additional details about the cause of the failure")
+ private FailureCause failureCause;
+
+ public FailureCause getFailureCause() {
+ return failureCause;
+ }
+
+ public void setFailureCause(FailureCause failureCause) {
+ this.failureCause = failureCause;
+ }
+
public String getDetails() {
return details;
}
@@ -76,12 +87,12 @@ public class C2OperationState implements Serializable {
return false;
}
C2OperationState that = (C2OperationState) o;
- return state == that.state && Objects.equals(details, that.details);
+ return state == that.state && Objects.equals(details, that.details) &&
Objects.equals(failureCause, that.failureCause);
}
@Override
public int hashCode() {
- return Objects.hash(state, details);
+ return Objects.hash(state, details, failureCause);
}
@Override
@@ -89,6 +100,7 @@ public class C2OperationState implements Serializable {
return "C2OperationState{" +
"state=" + state +
", details='" + details + '\'' +
+ ", failureCause='" + failureCause + '\'' +
'}';
}
diff --git
a/c2/c2-protocol/c2-protocol-api/src/main/java/org/apache/nifi/c2/protocol/api/FailureCause.java
b/c2/c2-protocol/c2-protocol-api/src/main/java/org/apache/nifi/c2/protocol/api/FailureCause.java
new file mode 100644
index 0000000000..56ef7ca1e4
--- /dev/null
+++
b/c2/c2-protocol/c2-protocol-api/src/main/java/org/apache/nifi/c2/protocol/api/FailureCause.java
@@ -0,0 +1,65 @@
+/*
+ * 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.c2.protocol.api;
+
+import java.util.List;
+import java.util.Objects;
+import org.apache.nifi.components.ValidationResult;
+
+public class FailureCause {
+ private List<ValidationResult> validationResults;
+ private String exceptionMessage;
+ private List<String> causedByMessages;
+
+ public List<ValidationResult> getValidationResults() {
+ return validationResults;
+ }
+
+ public void setValidationResults(List<ValidationResult> validationResults)
{
+ this.validationResults = validationResults;
+ }
+
+ public String getExceptionMessage() {
+ return exceptionMessage;
+ }
+
+ public void setExceptionMessage(String exceptionMessage) {
+ this.exceptionMessage = exceptionMessage;
+ }
+
+ public List<String> getCausedByMessages() {
+ return causedByMessages;
+ }
+
+ public void setCausedByMessages(List<String> causedByMessages) {
+ this.causedByMessages = causedByMessages;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ FailureCause that = (FailureCause) o;
+ return Objects.equals(validationResults, that.validationResults) &&
Objects.equals(exceptionMessage, that.exceptionMessage) &&
Objects.equals(causedByMessages, that.causedByMessages);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(validationResults, exceptionMessage,
causedByMessages);
+ }
+}
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 47504750be..7c1e7bddc5 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
@@ -54,6 +54,7 @@ import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.minifi.commons.service.FlowEnrichService;
import org.apache.nifi.minifi.commons.service.FlowPropertyEncryptor;
import org.apache.nifi.minifi.commons.service.FlowSerDeService;
+import org.apache.nifi.minifi.validator.ValidationException;
import org.apache.nifi.services.FlowService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -91,7 +92,7 @@ public class DefaultUpdateConfigurationStrategy implements
UpdateConfigurationSt
}
@Override
- public boolean update(byte[] rawFlow) {
+ public void update(byte[] rawFlow) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Attempting to update flow with content: \n{}", new
String(rawFlow, UTF_8));
}
@@ -116,22 +117,24 @@ public class DefaultUpdateConfigurationStrategy
implements UpdateConfigurationSt
reloadFlow(findAllProposedConnectionIds(enrichedFlowCandidate.getRootGroup()));
- return true;
} catch (IllegalStateException e) {
LOGGER.error("Configuration update failed. Reverting and reloading
previous flow", e);
revert(backupFlowConfigurationFile, flowConfigurationFile);
revert(backupRawFlowConfigurationFile, rawFlowConfigurationFile);
try {
reloadFlow(originalConnectionIds);
- } catch (IOException ex) {
- LOGGER.error("Unable to reload the reverted flow", e);
+ } catch (ValidationException ex) {
+ LOGGER.error("Unable to reload the reverted flow", ex);
+ throw ex;
+ } catch (Exception exception) {
+ throw new RuntimeException(exception);
}
- return false;
+ throw e;
} catch (Exception e) {
LOGGER.error("Configuration update failed. Reverting to previous
flow, no reload is necessary", e);
revert(backupFlowConfigurationFile, flowConfigurationFile);
revert(backupRawFlowConfigurationFile, rawFlowConfigurationFile);
- return false;
+ throw new RuntimeException(e);
} finally {
removeIfExists(backupFlowConfigurationFile);
removeIfExists(backupRawFlowConfigurationFile);
@@ -148,7 +151,7 @@ public class DefaultUpdateConfigurationStrategy implements
UpdateConfigurationSt
List<ValidationResult> validationErrors =
validate(flowController.getFlowManager());
if (!validationErrors.isEmpty()) {
LOGGER.error("Validation errors found when reloading the flow:
{}", validationErrors);
- throw new IllegalStateException("Unable to start flow due to
validation errors");
+ throw new ValidationException("Unable to start flow due to
validation errors", validationErrors);
}
flowController.getFlowManager().getRootGroup().startProcessing();
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/DefaultSyncResourceStrategy.java
b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategy.java
index 7c1f497192..427ab573a6 100644
---
a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategy.java
+++
b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategy.java
@@ -73,7 +73,7 @@ public class DefaultSyncResourceStrategy implements
SyncResourceStrategy {
@Override
public OperationState synchronizeResourceRepository(ResourcesGlobalHash
c2GlobalHash, List<ResourceItem> c2ServerItems,
BiFunction<String,
Function<InputStream, Optional<Path>>, Optional<Path>> resourceDownloadFunction,
- Function<String,
Optional<String>> urlEnrichFunction) {
+ Function<String,
String> urlEnrichFunction) {
Set<ResourceItem> c2Items = Set.copyOf(c2ServerItems);
Set<ResourceItem> agentItems =
Set.copyOf(resourceRepository.findAllResourceItems());
@@ -89,7 +89,7 @@ public class DefaultSyncResourceStrategy implements
SyncResourceStrategy {
private OperationState saveNewItems(Set<ResourceItem> c2Items,
Set<ResourceItem> agentItems,
BiFunction<String,
Function<InputStream, Optional<Path>>, Optional<Path>> resourceDownloadFunction,
- Function<String, Optional<String>>
urlEnrichFunction) {
+ Function<String, String>
urlEnrichFunction) {
List<ResourceItem> newItems =
c2Items.stream().filter(not(agentItems::contains)).toList();
if (newItems.isEmpty()) {
return NO_OPERATION;
@@ -107,10 +107,10 @@ public class DefaultSyncResourceStrategy implements
SyncResourceStrategy {
}
private Function<ResourceItem, Optional<ResourceItem>>
downloadIfNotPresentAndAddToRepository(
- BiFunction<String, Function<InputStream, Optional<Path>>,
Optional<Path>> resourceDownloadFunction, Function<String, Optional<String>>
urlEnrichFunction) {
+ BiFunction<String, Function<InputStream, Optional<Path>>,
Optional<Path>> resourceDownloadFunction, Function<String, String>
urlEnrichFunction) {
return resourceItem ->
resourceRepository.resourceItemBinaryPresent(resourceItem)
? resourceRepository.addResourceItem(resourceItem)
- : urlEnrichFunction.apply(resourceItem.getUrl())
+ :
Optional.ofNullable(urlEnrichFunction.apply(resourceItem.getUrl()))
.flatMap(enrichedUrl ->
resourceDownloadFunction.apply(enrichedUrl, this::persistToTemporaryLocation))
.flatMap(tempResourcePath ->
resourceRepository.addResourceItem(resourceItem, tempResourcePath));
}
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 87edfead9c..6c2226f803 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
@@ -127,10 +127,9 @@ public class DefaultUpdateConfigurationStrategyTest {
when(mockFlowManager.getRootGroup()).thenReturn(mockProcessGroup);
// when
- boolean result =
testUpdateConfigurationStrategy.update(NEW_RAW_FLOW_CONFIG_CONTENT);
+ testUpdateConfigurationStrategy.update(NEW_RAW_FLOW_CONFIG_CONTENT);
//then
- assertTrue(result);
assertTrue(exists(flowConfigurationFile));
assertTrue(exists(rawFlowConfigurationFile));
assertArrayEquals(NEW_ENRICHED_FLOW_CONFIG_CONTENT,
readGzipFile(flowConfigurationFile));
@@ -155,19 +154,20 @@ public class DefaultUpdateConfigurationStrategyTest {
doThrow(new IOException()).when(mockFlowService).load(null);
// when
- boolean result =
testUpdateConfigurationStrategy.update(NEW_RAW_FLOW_CONFIG_CONTENT);
-
- //then
- assertFalse(result);
- assertTrue(exists(flowConfigurationFile));
- assertTrue(exists(rawFlowConfigurationFile));
- assertArrayEquals(ORIGINAL_ENRICHED_FLOW_CONFIG_CONTENT,
readGzipFile(flowConfigurationFile));
- assertArrayEquals(ORIGINAL_RAW_FLOW_CONFIG_CONTENT,
readPlainTextFile(rawFlowConfigurationFile));
- assertFalse(exists(backupFlowConfigurationFile));
- assertFalse(exists(backupRawFlowConfigurationFile));
- verify(mockFlowService, times(1)).load(null);
- verify(mockFlowController, times(0)).onFlowInitialized(true);
- verify(mockProcessGroup, times(0)).startProcessing();
+ try {
+
testUpdateConfigurationStrategy.update(NEW_RAW_FLOW_CONFIG_CONTENT);
+ } catch (Exception e) {
+ //then
+ assertTrue(exists(flowConfigurationFile));
+ assertTrue(exists(rawFlowConfigurationFile));
+ assertArrayEquals(ORIGINAL_ENRICHED_FLOW_CONFIG_CONTENT,
readGzipFile(flowConfigurationFile));
+ assertArrayEquals(ORIGINAL_RAW_FLOW_CONFIG_CONTENT,
readPlainTextFile(rawFlowConfigurationFile));
+ assertFalse(exists(backupFlowConfigurationFile));
+ assertFalse(exists(backupRawFlowConfigurationFile));
+ verify(mockFlowService, times(1)).load(null);
+ verify(mockFlowController, times(0)).onFlowInitialized(true);
+ verify(mockProcessGroup, times(0)).startProcessing();
+ }
}
private void writeGzipFile(Path path, byte[] content) throws IOException {
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/DefaultSyncResourceStrategyTest.java
b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategyTest.java
index 1ba84d747d..e59eeed402 100644
---
a/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategyTest.java
+++
b/minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategyTest.java
@@ -70,7 +70,7 @@ public class DefaultSyncResourceStrategyTest {
(url, persistFunction) -> url.endsWith(FAIL_DOWNLOAD_URL) ? empty() :
persistFunction.apply(new ByteArrayInputStream(url.getBytes()));
private static String ENRICH_PREFIX = "pre_";
- private static final Function<String, Optional<String>>
PREFIXING_ENRICH_FUNCTION = url -> ofNullable(url).map(arg -> ENRICH_PREFIX +
arg);
+ private static final Function<String, String> PREFIXING_ENRICH_FUNCTION =
url -> ofNullable(url).map(arg -> ENRICH_PREFIX + arg).orElse("");
@Mock
private ResourceRepository mockResourceRepository;
