This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 585abb8432 fix: preserve pipelines on extension outages (#5181)
585abb8432 is described below
commit 585abb843251262f8814eb3176699c8762d08176
Author: Jacqueline Höllig <[email protected]>
AuthorDate: Mon Nov 2 19:15:32 2026 +0100
fix: preserve pipelines on extension outages (#5181)
Co-authored-by: Sven Oehler <[email protected]>
---
.../model/connect/adapter/PipelineUpdateInfo.java | 9 ++
.../CustomTransformOutputSchemaGenerator.java | 44 +++++--
.../output/OutputSchemaUnavailableException.java | 36 ++++++
.../update/DataStreamUpdateManagement.java | 2 +-
.../pipeline/update/PipelineUpdateCoordinator.java | 102 ++++++++++++++-
.../CustomTransformOutputSchemaGeneratorTest.java | 101 +++++++++++++++
.../update/DataStreamUpdateManagementTest.java | 24 +++-
.../update/PipelineUpdateCoordinatorTest.java | 59 +++++++++
.../streampipes/rest/impl/PipelineResource.java | 8 ++
ui/deployment/i18n/de.json | 4 +-
ui/deployment/i18n/en.json | 4 +-
ui/deployment/i18n/pl.json | 4 +-
.../src/lib/model/gen/streampipes-model.ts | 2 +
...adapter-started-update-migration.component.html | 28 ++++-
.../components/pipeline/pipeline.component.spec.ts | 64 ++++++++++
.../components/pipeline/pipeline.component.ts | 137 ++++++++++++++++-----
.../app/editor/services/pipeline-style.service.ts | 26 ++++
.../editor/services/pipeline-validation.service.ts | 25 +++-
18 files changed, 619 insertions(+), 60 deletions(-)
diff --git
a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/PipelineUpdateInfo.java
b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/PipelineUpdateInfo.java
index 8ee7184720..d77bd240ba 100644
---
a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/PipelineUpdateInfo.java
+++
b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/PipelineUpdateInfo.java
@@ -33,6 +33,7 @@ public class PipelineUpdateInfo {
private String pipelineName;
private boolean canAutoMigrate;
private String migrationInfo;
+ private boolean validationUnavailable;
private Map<String, List<PipelineElementValidationInfo>> validationInfos;
private List<ChartSchemaUpdateInfo> chartSchemaUpdateInfos;
@@ -73,6 +74,14 @@ public class PipelineUpdateInfo {
this.migrationInfo = migrationInfo;
}
+ public boolean isValidationUnavailable() {
+ return validationUnavailable;
+ }
+
+ public void setValidationUnavailable(boolean validationUnavailable) {
+ this.validationUnavailable = validationUnavailable;
+ }
+
public Map<String, List<PipelineElementValidationInfo>> getValidationInfos()
{
return validationInfos;
}
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 c9ee72c879..cc0a04812c 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;
@@ -79,8 +80,14 @@ public class CustomTransformOutputSchemaGenerator extends
OutputSchemaGenerator<
}
private EventSchema makeRequest() {
+ String httpRequestBody;
+ try {
+ httpRequestBody = makeRequestBody();
+ } catch (IOException e) {
+ throw new IllegalStateException("Could not create the output schema
request", e);
+ }
+
try {
- String httpRequestBody = makeRequestBody();
var service = new
ExtensionsServiceEndpointGenerator(serviceDiscovery).selectService(
dataProcessorInvocation.getAppId(),
SpServiceUrlProvider.DATA_PROCESSOR,
@@ -93,9 +100,8 @@ public class CustomTransformOutputSchemaGenerator extends
OutputSchemaGenerator<
);
var response =
requestManager.request(ExtensionServiceRequests.outputSchema(requestTarget,
httpRequestBody));
return handleResponse(response);
- } catch (Exception e) {
- e.printStackTrace();
- return new EventSchema();
+ } catch (NoServiceEndpointsAvailableException | IOException e) {
+ throw new
OutputSchemaUnavailableException(dataProcessorInvocation.getAppId(), e);
}
}
@@ -105,15 +111,33 @@ 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) {
+ throw new IllegalStateException("The extension returned no output schema
response");
+ }
+
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);
+ }
+ throw new IllegalStateException(cause);
}
- String resp = response.responseBody();
+ try {
+ var eventSchema = JacksonSerializer
+ .getObjectMapper()
+ .readValue(response.responseBody(), EventSchema.class);
+ if (eventSchema == null) {
+ throw new IllegalStateException("The extension returned no output
schema");
+ }
+ return eventSchema;
+ } catch (IOException | JsonSyntaxException e) {
+ throw new IllegalStateException("The extension returned an invalid
output schema", e);
+ }
+ }
- 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/matching/output/OutputSchemaUnavailableException.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/OutputSchemaUnavailableException.java
new file mode 100644
index 0000000000..498d08ca8c
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/OutputSchemaUnavailableException.java
@@ -0,0 +1,36 @@
+/*
+ * 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.streampipes.manager.matching.output;
+
+/**
+ * Signals that a custom output schema could not be determined from its
extension service.
+ */
+public class OutputSchemaUnavailableException extends RuntimeException {
+
+ private final String appId;
+
+ public OutputSchemaUnavailableException(String appId, Throwable cause) {
+ super("Could not compute the output schema for pipeline element " + appId,
cause);
+ this.appId = appId;
+ }
+
+ public String getAppId() {
+ return appId;
+ }
+}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagement.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagement.java
index 5de6578114..6ca7f87e06 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagement.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagement.java
@@ -48,8 +48,8 @@ public class DataStreamUpdateManagement {
boolean schemaChanged = EventSchemaChangeDetector.hasChanged(
previous == null ? null : previous.getEventSchema(),
dataStream.getEventSchema());
dataStreamResourceManager.update(dataStream);
- pipelineUpdateCoordinator.updatePipelines(dataStream);
if (schemaChanged) {
+ pipelineUpdateCoordinator.updatePipelines(dataStream);
eventPublisher.publishEvent(new
DataStreamSchemaUpdatedEvent(dataStream.getElementId(), dataStream.getName()));
}
}
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 7405727c07..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
@@ -22,12 +22,15 @@ import
org.apache.streampipes.commons.prometheus.pipelines.PipelinesStats;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.execution.PipelineExecutor;
import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
+import
org.apache.streampipes.manager.matching.output.OutputSchemaUnavailableException;
import
org.apache.streampipes.manager.matching.v2.pipeline.MeasurementChangeValidationStep;
import org.apache.streampipes.manager.pipeline.PipelineElementUserCleaner;
import org.apache.streampipes.manager.pipeline.PipelineManager;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
import org.apache.streampipes.model.connect.adapter.PipelineUpdateInfo;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.model.message.PipelineModificationMessage;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineElementValidationInfo;
@@ -43,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;
@@ -90,12 +95,10 @@ public class PipelineUpdateCoordinator {
affectedPipelines.forEach(pipeline -> {
var shouldRestartPipeline = pipeline.isRunning();
- if (shouldRestartPipeline) {
- new PipelineExecutor(pipeline, requestManager,
resourceManager).stopPipeline(true);
- }
-
+ var pipelineStopped = new AtomicBoolean();
var storedPipeline =
pipelineManager.getPipeline(pipeline.getPipelineId());
- var updatedPipeline = updatePipeline(storedPipeline, affectedElementId,
updatedStreamName, updatedEventSchema);
+ var updatedPipeline = updatePipeline(
+ copyPipeline(storedPipeline), affectedElementId, updatedStreamName,
updatedEventSchema);
try {
var verificationHandler = new
PipelineVerificationHandlerV2(resourceManager.getServiceDiscovery(),
updatedPipeline, requestManager);
@@ -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)
@@ -119,6 +124,12 @@ public class PipelineUpdateCoordinator {
modifiedPipeline.setValid(false);
}
+ if (shouldRestartPipeline) {
+ new PipelineExecutor(pipeline, requestManager,
resourceManager).stopPipeline(true);
+ pipelineStopped.set(true);
+ modifiedPipeline.setRunning(false);
+
modifiedPipeline.setRev(pipelineManager.getPipeline(pipeline.getPipelineId()).getRev());
+ }
PipelineElementUserCleaner.clearCorrespondingUsers(modifiedPipeline);
pipelineStorage.updateElement(modifiedPipeline);
@@ -127,7 +138,18 @@ public class PipelineUpdateCoordinator {
pipelineManager.getPipeline(pipeline.getPipelineId()),
requestManager, resourceManager
).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);
}
});
@@ -140,7 +162,8 @@ public class PipelineUpdateCoordinator {
var pipelineUpdateInfos = new ArrayList<PipelineUpdateInfo>();
affectedPipelines.forEach(pipeline -> {
- var updatedPipeline = updatePipeline(pipeline, affectedElementId,
updatedStreamName, updatedEventSchema);
+ var updatedPipeline = updatePipeline(
+ copyPipeline(pipeline), affectedElementId, updatedStreamName,
updatedEventSchema);
try {
var verificationHandler = new
PipelineVerificationHandlerV2(resourceManager.getServiceDiscovery(),
updatedPipeline, requestManager);
var modificationMessage = verificationHandler.verifyPipeline();
@@ -148,6 +171,8 @@ public class PipelineUpdateCoordinator {
updateInfo.setChartSchemaUpdateInfos(
chartSchemaUpdateCoordinator.checkChartMigrations(updatedPipeline,
updatedEventSchema));
pipelineUpdateInfos.add(updateInfo);
+ } catch (OutputSchemaUnavailableException e) {
+ pipelineUpdateInfos.add(makeUnavailableUpdateInfo(pipeline, e));
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -156,6 +181,71 @@ public class PipelineUpdateCoordinator {
return pipelineUpdateInfos;
}
+ private Pipeline copyPipeline(Pipeline pipeline) {
+ var copy = pipeline.clone();
+ copy.setRunning(pipeline.isRunning());
+ copy.setRestartOnSystemReboot(pipeline.isRestartOnSystemReboot());
+ copy.setStartedAt(pipeline.getStartedAt());
+ copy.setPublicElement(pipeline.isPublicElement());
+ copy.setLabels(new ArrayList<>(pipeline.getLabels()));
+ copy.setLastMigratedAt(pipeline.getLastMigratedAt());
+ copy.setPipelineNotifications(new
ArrayList<>(pipeline.getPipelineNotifications()));
+
copy.setStreams(pipeline.getStreams().stream().map(SpDataStream::new).toList());
+
copy.setSepas(pipeline.getSepas().stream().map(this::copyProcessor).toList());
+
copy.setActions(pipeline.getActions().stream().map(this::copySink).toList());
+ return copy;
+ }
+
+ private DataProcessorInvocation copyProcessor(DataProcessorInvocation
processor) {
+ var copy = new DataProcessorInvocation(processor);
+ copy.setSelectedServiceId(processor.getSelectedServiceId());
+ return copy;
+ }
+
+ private DataSinkInvocation copySink(DataSinkInvocation sink) {
+ var copy = new DataSinkInvocation(sink);
+ copy.setSelectedServiceId(sink.getSelectedServiceId());
+ return copy;
+ }
+
+ private PipelineUpdateInfo makeUnavailableUpdateInfo(Pipeline pipeline,
+
OutputSchemaUnavailableException exception) {
+ var updateInfo = new PipelineUpdateInfo();
+ updateInfo.setPipelineId(pipeline.getPipelineId());
+ updateInfo.setPipelineName(pipeline.getName());
+ updateInfo.setCanAutoMigrate(false);
+ updateInfo.setValidationUnavailable(true);
+ updateInfo.setMigrationInfo(exception.getMessage());
+ 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 a9986ab55c..0c256a82c7 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,12 +31,19 @@ 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.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 {
@@ -65,6 +74,98 @@ class CustomTransformOutputSchemaGeneratorTest {
assertTrue(secretProperty.getEncrypted());
}
+ @Test
+ void throwsWhenTheExtensionCannotComputeTheOutputSchema() {
+ var invocation = invocation();
+ var generator = new CustomTransformOutputSchemaGenerator(
+ mock(ISpServiceDiscovery.class),
+ new CustomTransformOutputStrategy(),
+ invocation,
+ mock(ExtensionServiceRequestManager.class)
+ );
+
+ var exception = assertThrows(OutputSchemaUnavailableException.class,
+ () -> generator.buildFromOneStream(null));
+
+ assertEquals(invocation.getAppId(), exception.getAppId());
+ }
+
+ @Test
+ void throwsWhenTheExtensionReturnsAnErrorResponse() 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 exception = assertThrows(IllegalStateException.class,
+ () -> generator.buildFromOneStream(null));
+
+ assertTrue(exception.getMessage().contains("status code: 500"));
+ }
+
+ @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 throwsWhenTheExtensionReturnsAnInvalidSchema() 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 exception = assertThrows(IllegalStateException.class,
+ () -> generator.buildFromOneStream(null));
+
+ assertTrue(exception.getMessage().contains("invalid output schema"));
+ }
+
+ 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/DataStreamUpdateManagementTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagementTest.java
index 97e16c7f0e..61bede7e8a 100644
---
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagementTest.java
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagementTest.java
@@ -61,7 +61,7 @@ class DataStreamUpdateManagementTest {
}
@Test
- void persistsStreamRenameWithoutSchemaEvent() {
+ void persistsStreamRenameWithoutUpdatingPipelinesOrPublishingSchemaEvent() {
var previous = stream();
var updated = new SpDataStream(previous);
updated.setName("Renamed stream");
@@ -76,8 +76,25 @@ class DataStreamUpdateManagementTest {
new DataStreamUpdateManagement(null, resources,
publisher).updateDataStream(updated);
verify(streams).update(updated);
- verify(coordinators.constructed().getFirst()).updatePipelines(updated);
- verifyNoInteractions(publisher);
+ verifyNoInteractions(publisher, coordinators.constructed().getFirst());
+ }
+ }
+
+ @Test
+ void
persistsUnchangedStreamWithoutUpdatingPipelinesOrPublishingSchemaEvent() {
+ var previous = stream();
+ var updated = new SpDataStream(previous);
+ var resources = mock(SpResourceManager.class);
+ var streams = mock(DataStreamResourceManager.class);
+ var publisher = mock(ApplicationEventPublisher.class);
+ when(resources.manageDataStreams()).thenReturn(streams);
+ when(streams.find("stream-1")).thenReturn(previous);
+
+ try (var coordinators = mockConstruction(PipelineUpdateCoordinator.class))
{
+ new DataStreamUpdateManagement(null, resources,
publisher).updateDataStream(updated);
+
+ verify(streams).update(updated);
+ verifyNoInteractions(publisher, coordinators.constructed().getFirst());
}
}
@@ -110,6 +127,7 @@ class DataStreamUpdateManagementTest {
var stream = new SpDataStream();
stream.setElementId("stream-1");
stream.setName("Stream");
+ stream.getEventSchema().addEventProperty(new
EventPropertyPrimitive("xs:string", "value", null, null));
return stream;
}
}
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 7927b61ebf..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
@@ -22,6 +22,7 @@ import org.apache.streampipes.audit.api.AuditService;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.execution.PipelineExecutor;
import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
+import
org.apache.streampipes.manager.matching.output.OutputSchemaUnavailableException;
import
org.apache.streampipes.manager.matching.v2.pipeline.MeasurementChangeValidationStep;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.connect.adapter.PipelineUpdateInfo;
@@ -190,6 +191,38 @@ class PipelineUpdateCoordinatorTest {
}
}
+ @Test
+ void
updatePipelines_ShouldDeferPipelineUpdateWhenOutputSchemaIsUnavailable() {
+ var requestManager = mock(ExtensionServiceRequestManager.class);
+ var resourceManager = mock(SpResourceManager.class);
+ var pipelineStorage = mock(IPipelineStorage.class);
+ var coordinator = makeCoordinator(requestManager, resourceManager,
pipelineStorage);
+ var dataStream = makeDataStream("stream-1", "Updated stream");
+ var storedPipeline = makePipeline("pipeline-1", "Pipeline", true,
"stream-1", "Old stream");
+ when(pipelineStorage.findAll()).thenReturn(List.of(storedPipeline));
+
when(pipelineStorage.getElementById("pipeline-1")).thenReturn(storedPipeline);
+
+ try (MockedConstruction<PipelineVerificationHandlerV2>
verificationHandlerConstruction =
+ mockConstruction(PipelineVerificationHandlerV2.class, (mock,
context) ->
+ when(mock.verifyPipeline()).thenThrow(new
OutputSchemaUnavailableException("math-expression",
+ new RuntimeException("Service unavailable"))));
+ MockedConstruction<PipelineExecutor> executorConstruction =
+ mockConstruction(PipelineExecutor.class)) {
+
+ coordinator.updatePipelines(dataStream);
+
+ assertEquals(1, verificationHandlerConstruction.constructed().size());
+ assertEquals(0, executorConstruction.constructed().size());
+ 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());
+ }
+ }
+
@Test
void checkPipelineMigrations_ShouldUseUpdatedDataStreamValues() {
var requestManager = mock(ExtensionServiceRequestManager.class);
@@ -262,6 +295,32 @@ class PipelineUpdateCoordinatorTest {
}
}
+ @Test
+ void
checkPipelineMigrations_ShouldReportUnavailableOutputSchemaWithoutMutatingThePipeline()
{
+ var requestManager = mock(ExtensionServiceRequestManager.class);
+ var resourceManager = mock(SpResourceManager.class);
+ var pipelineStorage = mock(IPipelineStorage.class);
+ var coordinator = makeCoordinator(requestManager, resourceManager,
pipelineStorage);
+ var dataStream = makeDataStream("stream-1", "Updated stream");
+ var pipeline = makePipeline("pipeline-1", "Pipeline", true, "stream-1",
"Old stream");
+ when(pipelineStorage.findAll()).thenReturn(List.of(pipeline));
+
+ try (MockedConstruction<PipelineVerificationHandlerV2>
verificationHandlerConstruction =
+ mockConstruction(PipelineVerificationHandlerV2.class, (mock,
context) ->
+ when(mock.verifyPipeline()).thenThrow(new
OutputSchemaUnavailableException("math-expression",
+ new RuntimeException("Service unavailable"))))) {
+
+ var result = coordinator.checkPipelineMigrations(dataStream);
+
+ assertEquals(1, verificationHandlerConstruction.constructed().size());
+ assertEquals(1, result.size());
+ assertTrue(result.getFirst().isValidationUnavailable());
+ assertFalse(result.getFirst().isCanAutoMigrate());
+
assertTrue(result.getFirst().getMigrationInfo().contains("math-expression"));
+ assertEquals("Old stream", pipeline.getStreams().getFirst().getName());
+ }
+ }
+
@Test
void
checkPipelineMigrations_ShouldDisableAutoMigrationForCriticalMeasurementFieldChange()
{
var requestManager = mock(ExtensionServiceRequestManager.class);
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
index d4513a9ad4..bd114638a9 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java
@@ -24,6 +24,7 @@ import
org.apache.streampipes.commons.exceptions.NoSuitableSepasAvailableExcepti
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.execution.status.PipelineStatusManager;
import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
+import
org.apache.streampipes.manager.matching.output.OutputSchemaUnavailableException;
import org.apache.streampipes.manager.pipeline.PipelineEditChanges;
import org.apache.streampipes.manager.pipeline.PipelineManager;
import
org.apache.streampipes.manager.pipeline.compact.CompactPipelineManagement;
@@ -286,6 +287,13 @@ public class PipelineResource extends
AbstractAuthGuardedRestResource {
public ResponseEntity<?> validatePipeline(@RequestBody Pipeline pipeline) {
try {
return ok(new
PipelineVerificationHandlerV2(resourceManager.getServiceDiscovery(), pipeline,
requestManager).verifyPipeline());
+ } catch (OutputSchemaUnavailableException e) {
+ LOG.warn("Could not validate pipeline because the output schema for {}
is unavailable", e.getAppId(), e);
+ return serverError(new Notification(
+ "Extension unavailable",
+ "Pipeline validation requires an extension that is currently
unavailable.",
+ e.getAppId()
+ ));
} catch (JsonSyntaxException e) {
return badRequest(new Notification(NotificationType.UNKNOWN_ERROR,
e.getMessage()));
} catch (Exception e) {
diff --git a/ui/deployment/i18n/de.json b/ui/deployment/i18n/de.json
index 466d1b4cdf..8adad43d32 100644
--- a/ui/deployment/i18n/de.json
+++ b/ui/deployment/i18n/de.json
@@ -138,7 +138,6 @@
"Assign additional labels to better discover your assets": "Weisen Sie
zusätzliche Labels zu, um Ihre Assets besser zu finden",
"Attach adapters, dashboards, or pipelines in the asset editor to surface
them here.": "Fügen Sie Adapter, Dashboards oder Pipelines im Asset-Editor
hinzu, um sie hier anzuzeigen.",
"Audit Log": "Audit-Protokoll",
- "Audit events could not be downloaded. Try again.": "Audit-Ereignisse
konnten nicht heruntergeladen werden. Versuche es erneut.",
"Audit events could not be loaded. Check the filters and storage
availability, then retry.": "Audit-Ereignisse konnten nicht geladen werden.
Prüfe die Filter und die Speicherverfügbarkeit und versuche es erneut.",
"Audit pages": "Audit-Seiten",
"Audit recording is disabled. Enable auditing in the Core deployment
configuration.": "Die Audit-Aufzeichnung ist deaktiviert. Aktiviere sie in der
Bereitstellungskonfiguration des Core.",
@@ -889,7 +888,6 @@
"Please wait while the pipeline is stopping": "Bitte warten Sie, während die
Pipeline gestoppt wird",
"Port": "Port",
"Possible placeholders:": "Mögliche Platzhalter:",
- "Preparing download...": "Download wird vorbereitet...",
"Preparing pipeline editor...": "Pipeline-Editor vorbereiten...",
"Press 'Next' to start the process.": "Auf \"Weiter\" drücken, um den
Vorgang zu starten.",
"Press Next to start the process.": "Klicken Sie auf Weiter, um den Vorgang
zu starten.",
@@ -1260,6 +1258,7 @@
"This operation cannot be undone.": "Dieser Vorgang kann nicht rückgängig
gemacht werden.",
"This operation cannot be undone. Please ensure that the data provider is
not used in a datalake retention.": "Dieser Vorgang kann nicht rückgängig
gemacht werden. Bitte stellen Sie sicher, dass der Provider in keiner
Speicherrichtlinie verwendet wird.",
"This pipeline cannot be updated because at least one data lake measurement
field changed its datatype.": "Diese Pipeline kann nicht aktualisiert werden,
da sich der Datentyp mindestens eines Data-Lake-Messfelds geändert hat.",
+ "This pipeline could not be revalidated because a required extension is
unavailable. Its configuration will remain unchanged.": "Diese Pipeline konnte
nicht erneut validiert werden, weil eine erforderliche Erweiterung nicht
verfügbar ist. Ihre Konfiguration bleibt unverändert.",
"This pipeline is still running.": "Diese Pipeline ist noch in Betrieb.",
"This pipeline update requires manual review.": "Diese
Pipeline-Aktualisierung erfordert eine manuelle Prüfung.",
"This pipeline will be automatically migrated.": "Diese Pipeline wird
automatisch migriert.",
@@ -1415,7 +1414,6 @@
"You need to delete those pipelines first before deleting the adapter.":
"Sie müssen diese Pipelines zuerst löschen, bevor Sie den Adapter löschen
können.",
"Your Assets were successfully added.": "Assets erfolgreich hinzugefügt.",
"Your Assets were successfully deleted.": "Assets erfolgreich gelöscht.",
- "Your audit log is being downloaded.": "Dein Audit-Protokoll wird
heruntergeladen.",
"Your new API key has been created. Please copy the key now - you won't be
able to see the key again.": "Ihr neuer API-Schlüssel wurde erstellt. Bitte
kopieren Sie den Schlüssel jetzt - Sie werden ihn nicht mehr sehen können.",
"Your new data stream is now available in the pipeline editor.": "Ihr neuer
Datenstrom ist nun im Pipeline-Editor verfügbar.",
"Your password successfully changed.": "Ihr Passwort wurde erfolgreich
geändert.",
diff --git a/ui/deployment/i18n/en.json b/ui/deployment/i18n/en.json
index 4a078bc89e..96d6c8b72e 100644
--- a/ui/deployment/i18n/en.json
+++ b/ui/deployment/i18n/en.json
@@ -138,7 +138,6 @@
"Assign additional labels to better discover your assets": null,
"Attach adapters, dashboards, or pipelines in the asset editor to surface
them here.": null,
"Audit Log": "Audit Log",
- "Audit events could not be downloaded. Try again.": null,
"Audit events could not be loaded. Check the filters and storage
availability, then retry.": "Audit events could not be loaded. Check the
filters and storage availability, then retry.",
"Audit pages": "Audit pages",
"Audit recording is disabled. Enable auditing in the Core deployment
configuration.": "Audit recording is disabled. Enable auditing in the Core
deployment configuration.",
@@ -889,7 +888,6 @@
"Please wait while the pipeline is stopping": null,
"Port": null,
"Possible placeholders:": null,
- "Preparing download...": null,
"Preparing pipeline editor...": null,
"Press 'Next' to start the process.": null,
"Press Next to start the process.": null,
@@ -1260,6 +1258,7 @@
"This operation cannot be undone.": null,
"This operation cannot be undone. Please ensure that the data provider is
not used in a datalake retention.": null,
"This pipeline cannot be updated because at least one data lake measurement
field changed its datatype.": null,
+ "This pipeline could not be revalidated because a required extension is
unavailable. Its configuration will remain unchanged.": null,
"This pipeline is still running.": null,
"This pipeline update requires manual review.": null,
"This pipeline will be automatically migrated.": null,
@@ -1415,7 +1414,6 @@
"You need to delete those pipelines first before deleting the adapter.":
null,
"Your Assets were successfully added.": null,
"Your Assets were successfully deleted.": null,
- "Your audit log is being downloaded.": null,
"Your new API key has been created. Please copy the key now - you won't be
able to see the key again.": null,
"Your new data stream is now available in the pipeline editor.": null,
"Your password successfully changed.": null,
diff --git a/ui/deployment/i18n/pl.json b/ui/deployment/i18n/pl.json
index 4ff0f740a8..7829c5162d 100644
--- a/ui/deployment/i18n/pl.json
+++ b/ui/deployment/i18n/pl.json
@@ -138,7 +138,6 @@
"Assign additional labels to better discover your assets": "Przypisz
dodatkowe etykiety, aby łatwiej odnajdywać zasoby",
"Attach adapters, dashboards, or pipelines in the asset editor to surface
them here.": "Dołącz adaptery, pulpity lub strumienie w edytorze zasobu, aby
wyświetlały się tutaj.",
"Audit Log": "Dziennik audytu",
- "Audit events could not be downloaded. Try again.": "Nie udało się pobrać
zdarzeń audytu. Spróbuj ponownie.",
"Audit events could not be loaded. Check the filters and storage
availability, then retry.": "Nie udało się wczytać zdarzeń audytu. Sprawdź
filtry i dostępność magazynu, a następnie spróbuj ponownie.",
"Audit pages": "Strony audytu",
"Audit recording is disabled. Enable auditing in the Core deployment
configuration.": "Rejestrowanie audytu jest wyłączone. Włącz je w konfiguracji
wdrożenia Core.",
@@ -889,7 +888,6 @@
"Please wait while the pipeline is stopping": "Proszę czekać, trwa
zatrzymywanie strumienia",
"Port": "Port",
"Possible placeholders:": "Możliwe zastępniki:",
- "Preparing download...": "Przygotowywanie pobierania...",
"Preparing pipeline editor...": "Przygotowywanie edytora strumieni...",
"Press 'Next' to start the process.": "Naciśnij „Dalej”, aby rozpocząć
proces.",
"Press Next to start the process.": "Kliknij Dalej, aby rozpocząć proces.",
@@ -1260,6 +1258,7 @@
"This operation cannot be undone.": "Tego działania nie można cofnąć.",
"This operation cannot be undone. Please ensure that the data provider is
not used in a datalake retention.": "Tego działania nie można cofnąć. Upewnij
się, że dostawca nie jest używany w retencji Data Lake.",
"This pipeline cannot be updated because at least one data lake measurement
field changed its datatype.": "Nie można zaktualizować tego strumienia,
ponieważ co najmniej jedno pole miary Data Lake zmieniło typ danych.",
+ "This pipeline could not be revalidated because a required extension is
unavailable. Its configuration will remain unchanged.": "Nie można było
ponownie zweryfikować tego strumienia, ponieważ wymagana wtyczka jest
niedostępna. Jego konfiguracja pozostanie bez zmian.",
"This pipeline is still running.": "Ten strumień nadal działa.",
"This pipeline update requires manual review.": "Ta aktualizacja strumienia
wymaga ręcznego sprawdzenia.",
"This pipeline will be automatically migrated.": "Ten strumień zostanie
automatycznie zmigrowany.",
@@ -1415,7 +1414,6 @@
"You need to delete those pipelines first before deleting the adapter.":
"Przed usunięciem adaptera najpierw usuń te strumienie.",
"Your Assets were successfully added.": "Twoje zasoby zostały pomyślnie
dodane.",
"Your Assets were successfully deleted.": "Twoje zasoby zostały pomyślnie
usunięte.",
- "Your audit log is being downloaded.": "Twój dziennik audytu jest
pobierany.",
"Your new API key has been created. Please copy the key now - you won't be
able to see the key again.": "Twój nowy klucz API został utworzony. Skopiuj go
teraz - później nie będzie już widoczny.",
"Your new data stream is now available in the pipeline editor.": "Twój nowy
strumień danych jest teraz dostępny w edytorze strumieńów.",
"Your password successfully changed.": "Twoje hasło zostało pomyślnie
zmienione.",
diff --git
a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index 92cf0d4c2e..85075aed61 100644
---
a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++
b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -3316,6 +3316,7 @@ export class PipelineUpdateInfo {
pipelineId: string;
pipelineName: string;
validationInfos: { [index: string]: PipelineElementValidationInfo[] };
+ validationUnavailable: boolean;
static fromData(
data: PipelineUpdateInfo,
@@ -3335,6 +3336,7 @@ export class PipelineUpdateInfo {
instance.validationInfos = __getCopyObjectFn(
__getCopyArrayFn(PipelineElementValidationInfo.fromData),
)(data.validationInfos);
+ instance.validationUnavailable = data.validationUnavailable;
return instance;
}
}
diff --git
a/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html
b/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html
index 4775d0f3db..b13f513426 100644
---
a/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html
+++
b/ui/src/app/connect/dialog/adapter-started/adapter-started-update-migration/adapter-started-update-migration.component.html
@@ -61,7 +61,33 @@
</div>
</div>
}
- @if (!updateInfo.canAutoMigrate) {
+ @if (updateInfo.validationUnavailable) {
+ <div
+ class="pipeline-status"
+ fxLayoutAlign="start center"
+
data-cy="adapter-pipeline-validation-unavailable-warning"
+ >
+ <mat-icon class="status-icon status-icon-warning">
+ warning
+ </mat-icon>
+ <div>
+ <div class="pipeline-title">
+ {{ 'Pipeline' | translate }}
+ <b>{{ updateInfo.pipelineName }}</b>
+ </div>
+ <div class="pipeline-description">
+ {{
+ 'This pipeline could not be
revalidated because a required extension is unavailable. Its configuration will
remain unchanged.'
+ | translate
+ }}
+ </div>
+ </div>
+ </div>
+ }
+ @if (
+ !updateInfo.canAutoMigrate &&
+ !updateInfo.validationUnavailable
+ ) {
<div
class="pipeline-status"
fxLayoutAlign="start center"
diff --git a/ui/src/app/editor/components/pipeline/pipeline.component.spec.ts
b/ui/src/app/editor/components/pipeline/pipeline.component.spec.ts
index 92f5db4f00..9107ab64f6 100644
--- a/ui/src/app/editor/components/pipeline/pipeline.component.spec.ts
+++ b/ui/src/app/editor/components/pipeline/pipeline.component.spec.ts
@@ -17,6 +17,7 @@
*/
import { TestBed } from '@angular/core/testing';
+import { HttpErrorResponse } from '@angular/common/http';
import { MatDialog } from '@angular/material/dialog';
import { DialogService, KeyboardShortcutService } from
'@streampipes/shared-ui';
import { DataProcessorInvocation } from '@streampipes/platform-services';
@@ -30,6 +31,7 @@ import { EditorService } from '../../services/editor.service';
import { PipelineStyleService } from '../../services/pipeline-style.service';
import { PipelineValidationService } from
'../../services/pipeline-validation.service';
import { IdGeneratorService } from
'../../../core-services/id-generator/id-generator.service';
+import { PipelineElementConfigurationStatus } from '../../model/editor.model';
describe('Pipeline palette drops', () => {
const create = vi.fn();
@@ -146,4 +148,66 @@ describe('Pipeline palette drops', () => {
});
expect(create).not.toHaveBeenCalled();
});
+
+ it('marks an unavailable extension and its connections as invalid', () => {
+ const { component } = setup();
+ const unavailableExtension = {
+ type: 'sepa',
+ payload: Object.assign(new DataProcessorInvocation(), {
+ appId:
'org.apache.streampipes.processors.enricher.jvm.processor.expression',
+ dom: 'math-expression',
+ name: 'Math Expression',
+ }),
+ settings: {
+ completed: 1,
+ },
+ } as PipelineComponent['rawPipelineModel'][number];
+ component.rawPipelineModel = [unavailableExtension];
+
+ const pipelineStyleService = component[
+ 'pipelineStyleService'
+ ] as unknown as {
+ updatePeConfigurationStatus: ReturnType<typeof vi.fn>;
+ updateConnectedConnectorStylesAsError: ReturnType<typeof vi.fn>;
+ };
+ pipelineStyleService.updatePeConfigurationStatus = vi.fn();
+ pipelineStyleService.updateConnectedConnectorStylesAsError = vi.fn();
+ const validationService = component[
+ 'pipelineValidationService'
+ ] as unknown as {
+ reportRemoteValidationError: ReturnType<typeof vi.fn>;
+ };
+ validationService.reportRemoteValidationError = vi.fn();
+
+ component['handlePipelineValidationFailure'](
+ new HttpErrorResponse({
+ error: {
+ title: 'Extension unavailable',
+ description:
+ 'Pipeline validation requires an extension that is
currently unavailable.',
+ additionalInformation:
+
'org.apache.streampipes.processors.enricher.jvm.processor.expression',
+ },
+ }),
+ );
+
+ expect(
+ pipelineStyleService.updatePeConfigurationStatus,
+ ).toHaveBeenCalledWith(
+ unavailableExtension,
+ PipelineElementConfigurationStatus.INCOMPLETE,
+ );
+ expect(
+ pipelineStyleService.updateConnectedConnectorStylesAsError,
+ ).toHaveBeenCalledWith('math-expression');
+ expect(
+ validationService.reportRemoteValidationError,
+ ).toHaveBeenCalledWith(
+ 'Extension unavailable',
+ 'Pipeline validation requires an extension that is currently
unavailable. (Math Expression
(org.apache.streampipes.processors.enricher.jvm.processor.expression))',
+ );
+ expect(
+ component.JsplumbBridge.repaintEverything,
+ ).toHaveBeenCalledOnce();
+ });
});
diff --git a/ui/src/app/editor/components/pipeline/pipeline.component.ts
b/ui/src/app/editor/components/pipeline/pipeline.component.ts
index 2ee290631e..ffb9b41285 100644
--- a/ui/src/app/editor/components/pipeline/pipeline.component.ts
+++ b/ui/src/app/editor/components/pipeline/pipeline.component.ts
@@ -56,7 +56,6 @@ import {
import { ObjectProvider } from '../../services/object-provider.service';
import { CustomizeComponent } from
'../../dialog/customize/customize.component';
import {
- ConfirmDialogComponent,
DialogService,
KeyboardShortcutService,
PanelType,
@@ -64,7 +63,6 @@ import {
} from '@streampipes/shared-ui';
import { EditorService } from '../../services/editor.service';
import { MatchingErrorComponent } from
'../../dialog/matching-error/matching-error.component';
-import { MatDialog } from '@angular/material/dialog';
import { JsplumbFactoryService } from '../../services/jsplumb-factory.service';
import {
EVENT_CONNECTION,
@@ -75,6 +73,7 @@ import {
} from '@jsplumb/browser-ui';
import { PipelineStyleService } from '../../services/pipeline-style.service';
import { IdGeneratorService } from
'../../../core-services/id-generator/id-generator.service';
+import { HttpErrorResponse } from '@angular/common/http';
import { NgStyle } from '@angular/common';
import { StyleDirective } from '@ngbracket/ngx-layout/extended';
import { DroppedPipelineElementComponent } from
'./dropped-pipeline-element/dropped-pipeline-element.component';
@@ -112,7 +111,6 @@ export class PipelineComponent implements OnInit, OnDestroy
{
private pipelineStyleService = inject(PipelineStyleService);
private pipelineValidationService = inject(PipelineValidationService);
private dialogService = inject(DialogService);
- private dialog = inject(MatDialog);
private ngZone = inject(NgZone);
private shortcutService = inject(KeyboardShortcutService);
@@ -154,6 +152,10 @@ export class PipelineComponent implements OnInit,
OnDestroy {
shouldOpenCustomizeSettings = false;
private shortcutReg: ShortcutRegistration;
+ private unavailablePipelineElementStatuses = new Map<
+ PipelineElementConfig,
+ PipelineElementConfigurationStatus
+ >();
constructor() {
this.currentPipelineModel = new Pipeline();
@@ -485,10 +487,7 @@ export class PipelineComponent implements OnInit,
OnDestroy {
},
status => {
pe.settings.loadingStatus = false;
- this.showErrorDialog(
- status.error.title,
- status.error.description,
- );
+ this.handlePipelineValidationFailure(status);
},
);
}
@@ -514,6 +513,7 @@ export class PipelineComponent implements OnInit, OnDestroy
{
}
modifyPipeline(pm: PipelineModificationMessage) {
+ this.restoreUnavailablePipelineElements();
if (pm.pipelineModifications) {
pm.pipelineModifications.forEach(modification => {
const id = modification.domId;
@@ -579,15 +579,90 @@ export class PipelineComponent implements OnInit,
OnDestroy {
return custom;
}
- showErrorDialog(title: string, description: string) {
- this.dialog.open(ConfirmDialogComponent, {
- width: '500px',
- data: {
- title: title,
- subtitle: description,
- confirmTitle: 'Ok',
+ private handlePipelineValidationFailure(status: HttpErrorResponse): void {
+ const error = status.error as Notification;
+ const title = error?.title ?? 'Pipeline validation failed';
+ const description = error?.description ?? status.message;
+ const extensionId = error?.additionalInformation;
+ const unavailableExtensions =
+ this.getUnavailableExtensions(extensionId);
+
+ if (title === 'Extension unavailable' && unavailableExtensions.length)
{
+ unavailableExtensions.forEach(pipelineElement => {
+ if (
+ !this.unavailablePipelineElementStatuses.has(
+ pipelineElement,
+ )
+ ) {
+ this.unavailablePipelineElementStatuses.set(
+ pipelineElement,
+ pipelineElement.settings.completed,
+ );
+ }
+ this.pipelineStyleService.updatePeConfigurationStatus(
+ pipelineElement,
+ PipelineElementConfigurationStatus.INCOMPLETE,
+ );
+
this.pipelineStyleService.updateConnectedConnectorStylesAsError(
+ pipelineElement.payload.dom,
+ );
+ });
+ this.JsplumbBridge.repaintEverything();
+ }
+
+ this.pipelineValid = false;
+ this.pipelineValidationService.reportRemoteValidationError(
+ title,
+ this.getUnavailableExtensionDescription(
+ description,
+ extensionId,
+ unavailableExtensions,
+ ),
+ );
+ }
+
+ private getUnavailableExtensions(
+ extensionId?: string,
+ ): PipelineElementConfig[] {
+ if (!extensionId) {
+ return [];
+ }
+
+ return this.rawPipelineModel.filter(
+ pipelineElement => pipelineElement.payload.appId === extensionId,
+ );
+ }
+
+ private getUnavailableExtensionDescription(
+ description: string,
+ extensionId: string | undefined,
+ unavailableExtensions: PipelineElementConfig[],
+ ): string {
+ if (!extensionId) {
+ return description;
+ }
+
+ const extensionNames = unavailableExtensions.map(
+ pipelineElement => pipelineElement.payload.name,
+ );
+ const unavailableExtension =
+ extensionNames.length > 0
+ ? `${extensionNames.join(', ')} (${extensionId})`
+ : extensionId;
+
+ return `${description} (${unavailableExtension})`;
+ }
+
+ private restoreUnavailablePipelineElements(): void {
+ this.unavailablePipelineElementStatuses.forEach(
+ (configurationStatus, pipelineElement) => {
+ this.pipelineStyleService.updatePeConfigurationStatus(
+ pipelineElement,
+ configurationStatus,
+ );
},
- });
+ );
+ this.unavailablePipelineElementStatuses.clear();
}
showMatchingErrorDialog(notifications: Notification[]) {
@@ -620,15 +695,20 @@ export class PipelineComponent implements OnInit,
OnDestroy {
);
this.objectProvider
.updatePipeline(this.currentPipelineModel)
- .subscribe(pm => {
- this.modifyPipeline(pm);
- this.triggerPipelineCacheUpdateEmitter.emit();
- this.announceConfiguredElement(pipelineElementConfig);
- if (this.previewModeActive) {
- this.deletePreviewEmitter.emit(true);
- }
- this.validatePipeline(pm);
- });
+ .subscribe(
+ pm => {
+ this.modifyPipeline(pm);
+ this.triggerPipelineCacheUpdateEmitter.emit();
+ this.announceConfiguredElement(
+ pipelineElementConfig,
+ );
+ if (this.previewModeActive) {
+ this.deletePreviewEmitter.emit(true);
+ }
+ this.validatePipeline(pm);
+ },
+ status => this.handlePipelineValidationFailure(status),
+ );
} else {
this.validatePipeline();
}
@@ -643,9 +723,8 @@ export class PipelineComponent implements OnInit, OnDestroy
{
this.currentPipelineModel = this.objectProvider.makePipeline(
this.rawPipelineModel,
);
- this.objectProvider
- .updatePipeline(this.currentPipelineModel)
- .subscribe(pm => {
+
this.objectProvider.updatePipeline(this.currentPipelineModel).subscribe(
+ pm => {
this.modifyPipeline(pm);
this.pipelineValid =
this.pipelineValidationService.isValidPipeline(
@@ -653,7 +732,9 @@ export class PipelineComponent implements OnInit, OnDestroy
{
this.readonly,
pm,
);
- });
+ },
+ status => this.handlePipelineValidationFailure(status),
+ );
}
openTopicsDialog(pipelineElementConfig: PipelineElementConfig) {
diff --git a/ui/src/app/editor/services/pipeline-style.service.ts
b/ui/src/app/editor/services/pipeline-style.service.ts
index 71da6b8344..df157cdce3 100644
--- a/ui/src/app/editor/services/pipeline-style.service.ts
+++ b/ui/src/app/editor/services/pipeline-style.service.ts
@@ -69,6 +69,32 @@ export class PipelineStyleService {
}
}
+ updateConnectedConnectorStylesAsError(pipelineElementDomId: string) {
+ const jsplumbBridge =
+ this.jsplumbFactoryService.getJsplumbBridge(false);
+ const pipelineElement = this.byId(pipelineElementDomId);
+ const sourceConnections = jsplumbBridge.getConnections({
+ source: pipelineElement,
+ });
+ const targetConnections = jsplumbBridge.getConnections({
+ target: pipelineElement,
+ });
+ const connections = [
+ ...(Array.isArray(sourceConnections)
+ ? sourceConnections
+ : Object.values(sourceConnections)),
+ ...(Array.isArray(targetConnections)
+ ? targetConnections
+ : Object.values(targetConnections)),
+ ];
+ const connectorStyle =
+ this.jsPlumbConfigService.getConnectorStyleError();
+
+ connections.forEach(connection => {
+ connection.setPaintStyle(connectorStyle);
+ });
+ }
+
getConnectorStyleConfig(status: EdgeValidationStatus) {
if (status.validationStatusType === 'COMPLETE') {
return this.jsPlumbConfigService.getConnectorStyleSuccess();
diff --git a/ui/src/app/editor/services/pipeline-validation.service.ts
b/ui/src/app/editor/services/pipeline-validation.service.ts
index f6eb6c97b3..98123a16bd 100644
--- a/ui/src/app/editor/services/pipeline-validation.service.ts
+++ b/ui/src/app/editor/services/pipeline-validation.service.ts
@@ -36,8 +36,9 @@ import { Connection } from '@jsplumb/browser-ui';
export class PipelineValidationService {
private jsplumbFactoryService = inject(JsplumbFactoryService);
- errorMessages: any = [];
+ errorMessages: UserErrorMessage[] = [];
pipelineValid = false;
+ private remoteValidationError?: UserErrorMessage;
availableErrorMessages: UserErrorMessage[] = [
new UserErrorMessage(
@@ -71,6 +72,9 @@ export class PipelineValidationService {
previewConfig: boolean,
pm?: PipelineModificationMessage,
) {
+ if (pm) {
+ this.remoteValidationError = undefined;
+ }
const jsplumbBridge =
this.jsplumbFactoryService.getJsplumbBridge(previewConfig);
const streamInAssembly = this.isStreamInAssembly(rawPipelineModel);
@@ -112,17 +116,34 @@ export class PipelineValidationService {
this.errorMessages.push(this.availableErrorMessages[5]);
}
+ if (this.remoteValidationError) {
+ this.errorMessages.push(this.remoteValidationError);
+ }
+
this.pipelineValid =
streamInAssembly &&
actionInAssembly &&
allElementsConnected &&
onlyOnePipelineCreated &&
allElementsConfigured &&
- (pm ? pm.pipelineValid : true);
+ (pm ? pm.pipelineValid : true) &&
+ !this.remoteValidationError;
return this.pipelineValid;
}
+ reportRemoteValidationError(title: string, content: string): void {
+ const previousRemoteValidationError = this.remoteValidationError;
+ this.remoteValidationError = new UserErrorMessage(title, content);
+ this.errorMessages = [
+ ...this.errorMessages.filter(
+ error => error !== previousRemoteValidationError,
+ ),
+ this.remoteValidationError,
+ ];
+ this.pipelineValid = false;
+ }
+
isEmptyPipeline(rawPipelineModel) {
return (
!this.isActionInAssembly(rawPipelineModel) &&
