This is an automated email from the ASF dual-hosted git repository.
JHoelli pushed a commit to branch marking_pipeline_falsely_as_rewuire_attention
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/marking_pipeline_falsely_as_rewuire_attention by this push:
new 073605ee2f minor
073605ee2f is described below
commit 073605ee2fcbd68792de2aea91f8508d1afb0221
Author: Jacqueline Höllig <[email protected]>
AuthorDate: Mon Nov 2 10:33:18 2026 +0100
minor
---
.../CustomTransformOutputSchemaGenerator.java | 33 ++++++--
.../pipeline/update/PipelineUpdateCoordinator.java | 41 ++++++++++
.../CustomTransformOutputSchemaGeneratorTest.java | 88 +++++++++++++++++++++-
.../update/PipelineUpdateCoordinatorTest.java | 8 +-
4 files changed, 158 insertions(+), 12 deletions(-)
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
index 8f8a3da94d..193f4ab98a 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
@@ -17,6 +17,7 @@
*/
package org.apache.streampipes.manager.matching.output;
+import
org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceOperationResult;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets;
@@ -93,8 +94,12 @@ public class CustomTransformOutputSchemaGenerator extends
OutputSchemaGenerator<
);
var response =
requestManager.request(ExtensionServiceRequests.outputSchema(requestTarget,
httpRequestBody));
return handleResponse(response);
- } catch (Exception e) {
+ } catch (OutputSchemaUnavailableException e) {
+ throw e;
+ } catch (NoServiceEndpointsAvailableException | IOException e) {
throw new
OutputSchemaUnavailableException(dataProcessorInvocation.getAppId(), e);
+ } catch (Exception e) {
+ return new EventSchema();
}
}
@@ -104,15 +109,29 @@ public class CustomTransformOutputSchemaGenerator extends
OutputSchemaGenerator<
return
JacksonSerializer.getObjectMapper().writeValueAsString(requestInvocation);
}
- private EventSchema handleResponse(ExtensionServiceOperationResult response)
throws JsonSyntaxException, IOException {
+ private EventSchema handleResponse(ExtensionServiceOperationResult response)
{
+ if (response == null) {
+ return new EventSchema();
+ }
+
if (!response.isSuccess()) {
- throw new IOException("Could not compute output schema, status code: " +
response.statusCode());
+ var cause = new IOException("Could not compute output schema, status
code: " + response.statusCode());
+ if (isUnavailableStatus(response.statusCode())) {
+ throw new
OutputSchemaUnavailableException(dataProcessorInvocation.getAppId(), cause);
+ }
+ return new EventSchema();
}
- String resp = response.responseBody();
+ try {
+ return JacksonSerializer
+ .getObjectMapper()
+ .readValue(response.responseBody(), EventSchema.class);
+ } catch (IOException | JsonSyntaxException e) {
+ return new EventSchema();
+ }
+ }
- return JacksonSerializer
- .getObjectMapper()
- .readValue(resp, EventSchema.class);
+ private boolean isUnavailableStatus(int statusCode) {
+ return statusCode == 502 || statusCode == 503 || statusCode == 504;
}
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
index 469fcbec46..6dadd74ab4 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
@@ -46,12 +46,14 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Stream;
public class PipelineUpdateCoordinator {
private static final Logger LOG =
LoggerFactory.getLogger(PipelineUpdateCoordinator.class);
private static final PipelinesStats PIPELINES_STATS = new PipelinesStats();
+ private static final String DEFERRED_VALIDATION_NOTIFICATION_PREFIX =
"Validation deferred: ";
private final ExtensionServiceRequestManager requestManager;
private final ChartSchemaUpdateCoordinator chartSchemaUpdateCoordinator;
@@ -93,6 +95,7 @@ public class PipelineUpdateCoordinator {
affectedPipelines.forEach(pipeline -> {
var shouldRestartPipeline = pipeline.isRunning();
+ var pipelineStopped = new AtomicBoolean();
var storedPipeline =
pipelineManager.getPipeline(pipeline.getPipelineId());
var updatedPipeline = updatePipeline(
copyPipeline(storedPipeline), affectedElementId, updatedStreamName,
updatedEventSchema);
@@ -104,6 +107,8 @@ public class PipelineUpdateCoordinator {
var modifiedPipeline =
verificationHandler.makeModifiedPipeline(modificationMessage).pipeline();
var canAutoMigrate = canAutoMigrate(modificationMessage);
+ removeDeferredValidationNotifications(modifiedPipeline);
+
if (!canAutoMigrate) {
modifiedPipeline.setHealthStatus(
requiresMeasurementUpdate(modificationMessage)
@@ -121,6 +126,7 @@ public class PipelineUpdateCoordinator {
if (shouldRestartPipeline) {
new PipelineExecutor(pipeline, requestManager,
resourceManager).stopPipeline(true);
+ pipelineStopped.set(true);
modifiedPipeline.setRunning(false);
modifiedPipeline.setRev(pipelineManager.getPipeline(pipeline.getPipelineId()).getRev());
}
@@ -133,9 +139,17 @@ public class PipelineUpdateCoordinator {
).startPipeline();
}
} catch (OutputSchemaUnavailableException e) {
+ try {
+ deferPipelineUpdate(storedPipeline, notificationType, e);
+ } catch (Exception notificationException) {
+ LOG.warn("Could not record the deferred pipeline update for {}",
storedPipeline.getName(), notificationException);
+ }
LOG.warn("Skipped pipeline update for {} because the output schema for
{} is unavailable",
storedPipeline.getName(), e.getAppId(), e);
} catch (Exception e) {
+ if (shouldRestartPipeline && !pipelineStopped.get()) {
+ new PipelineExecutor(pipeline, requestManager,
resourceManager).stopPipeline(true);
+ }
LOG.error("Could not update pipeline {}", updatedPipeline.getName(),
e);
}
});
@@ -205,6 +219,33 @@ public class PipelineUpdateCoordinator {
return updateInfo;
}
+ private void deferPipelineUpdate(Pipeline pipeline,
+ String notificationType,
+ OutputSchemaUnavailableException exception)
{
+ var notification = deferredValidationNotification(notificationType,
exception.getAppId());
+ var notifications = new ArrayList<>(pipeline.getPipelineNotifications());
+ if (notifications.contains(notification)) {
+ return;
+ }
+
+ notifications.add(notification);
+ pipeline.setPipelineNotifications(notifications);
+ pipelineStorage.updateElement(pipeline);
+ }
+
+ private void removeDeferredValidationNotifications(Pipeline pipeline) {
+
pipeline.setPipelineNotifications(pipeline.getPipelineNotifications().stream()
+ .filter(notification ->
!notification.startsWith(DEFERRED_VALIDATION_NOTIFICATION_PREFIX))
+ .toList());
+ }
+
+ private String deferredValidationNotification(String notificationType,
+ String appId) {
+ return DEFERRED_VALIDATION_NOTIFICATION_PREFIX
+ + "%s modification could not be validated because pipeline element %s
is unavailable".formatted(
+ notificationType, appId);
+ }
+
private Pipeline updatePipeline(Pipeline pipeline,
String affectedElementId,
String updatedStreamName,
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGeneratorTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGeneratorTest.java
index 2e35fbef98..bf54707813 100644
---
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGeneratorTest.java
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGeneratorTest.java
@@ -18,7 +18,9 @@
package org.apache.streampipes.manager.matching.output;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceOperationResult;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import
org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.output.CustomTransformOutputStrategy;
import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
@@ -29,13 +31,20 @@ import
org.apache.streampipes.user.management.encryption.SecretEncryptionManager
import org.junit.jupiter.api.Test;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
class CustomTransformOutputSchemaGeneratorTest {
@@ -68,8 +77,7 @@ class CustomTransformOutputSchemaGeneratorTest {
@Test
void throwsWhenTheExtensionCannotComputeTheOutputSchema() {
- var invocation = new DataProcessorInvocation();
- invocation.setAppId("org.apache.streampipes.test.processor");
+ var invocation = invocation();
var generator = new CustomTransformOutputSchemaGenerator(
mock(ISpServiceDiscovery.class),
new CustomTransformOutputStrategy(),
@@ -83,6 +91,82 @@ class CustomTransformOutputSchemaGeneratorTest {
assertEquals(invocation.getAppId(), exception.getAppId());
}
+ @Test
+ void returnsAnEmptySchemaWhenTheExtensionReturnsAnErrorResponse() throws
IOException {
+ var invocation = invocation();
+ var serviceDiscovery = mock(ISpServiceDiscovery.class);
+ var requestManager = mock(ExtensionServiceRequestManager.class);
+ when(serviceDiscovery.getService(anyString(), eq(true),
anyList())).thenReturn(List.of(service()));
+ when(requestManager.request(any())).thenReturn(new
ExtensionServiceOperationResult(500, null));
+ var generator = new CustomTransformOutputSchemaGenerator(
+ serviceDiscovery,
+ new CustomTransformOutputStrategy(),
+ invocation,
+ requestManager
+ );
+
+ var result = generator.buildFromOneStream(null);
+
+ assertNotNull(result.k);
+ assertTrue(result.k.getEventProperties().isEmpty());
+ }
+
+ @Test
+ void treatsGatewayUnavailableResponsesAsOutputSchemaUnavailable() throws
IOException {
+ var invocation = invocation();
+ var serviceDiscovery = mock(ISpServiceDiscovery.class);
+ var requestManager = mock(ExtensionServiceRequestManager.class);
+ when(serviceDiscovery.getService(anyString(), eq(true),
anyList())).thenReturn(List.of(service()));
+ when(requestManager.request(any())).thenReturn(new
ExtensionServiceOperationResult(503, null));
+ var generator = new CustomTransformOutputSchemaGenerator(
+ serviceDiscovery,
+ new CustomTransformOutputStrategy(),
+ invocation,
+ requestManager
+ );
+
+ var exception = assertThrows(OutputSchemaUnavailableException.class,
+ () -> generator.buildFromOneStream(null));
+
+ assertEquals(invocation.getAppId(), exception.getAppId());
+ }
+
+ @Test
+ void returnsAnEmptySchemaWhenTheExtensionReturnsAnInvalidSchema() throws
IOException {
+ var invocation = invocation();
+ var serviceDiscovery = mock(ISpServiceDiscovery.class);
+ var requestManager = mock(ExtensionServiceRequestManager.class);
+ when(serviceDiscovery.getService(anyString(), eq(true),
anyList())).thenReturn(List.of(service()));
+ when(requestManager.request(any())).thenReturn(new
ExtensionServiceOperationResult(
+ 200, "not-an-event-schema".getBytes(StandardCharsets.UTF_8)));
+ var generator = new CustomTransformOutputSchemaGenerator(
+ serviceDiscovery,
+ new CustomTransformOutputStrategy(),
+ invocation,
+ requestManager
+ );
+
+ var result = generator.buildFromOneStream(null);
+
+ assertNotNull(result.k);
+ assertTrue(result.k.getEventProperties().isEmpty());
+ }
+
+ private DataProcessorInvocation invocation() {
+ var invocation = new DataProcessorInvocation();
+ invocation.setAppId("org.apache.streampipes.test.processor");
+ invocation.setStaticProperties(List.of());
+ return invocation;
+ }
+
+ private SpServiceRegistration service() {
+ var service = new SpServiceRegistration();
+ service.setHost("localhost");
+ service.setPort(8080);
+ service.setSvcId("test-service");
+ return service;
+ }
+
private SecretStaticProperty makeSecretProperty(String value,
boolean encrypted) {
var secretProperty = new SecretStaticProperty("secret", "Secret", "");
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
index 1a610a1917..69753f5aff 100644
---
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinatorTest.java
@@ -58,7 +58,6 @@ import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
-import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -193,7 +192,7 @@ class PipelineUpdateCoordinatorTest {
}
@Test
- void updatePipelines_ShouldPreservePipelineWhenOutputSchemaIsUnavailable() {
+ void
updatePipelines_ShouldDeferPipelineUpdateWhenOutputSchemaIsUnavailable() {
var requestManager = mock(ExtensionServiceRequestManager.class);
var resourceManager = mock(SpResourceManager.class);
var pipelineStorage = mock(IPipelineStorage.class);
@@ -214,10 +213,13 @@ class PipelineUpdateCoordinatorTest {
assertEquals(1, verificationHandlerConstruction.constructed().size());
assertEquals(0, executorConstruction.constructed().size());
- verify(pipelineStorage, never()).updateElement(storedPipeline);
+ verify(pipelineStorage).updateElement(storedPipeline);
assertTrue(storedPipeline.isRunning());
assertTrue(storedPipeline.isValid());
assertEquals("Old stream",
storedPipeline.getStreams().getFirst().getName());
+ assertEquals(List.of(
+ "Validation deferred: Data stream modification could not be
validated because pipeline element math-expression is unavailable"
+ ), storedPipeline.getPipelineNotifications());
}
}
