This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch fix-adapter-producer-lifecycle
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/fix-adapter-producer-lifecycle
by this push:
new a12a7555f3 Clean up pipeline elements on failed start/stop
a12a7555f3 is described below
commit a12a7555f313e656d6df8e70d54ada8c1d39c1ca
Author: Sven Oehler <[email protected]>
AuthorDate: Tue Jul 14 14:54:04 2026 +0200
Clean up pipeline elements on failed start/stop
---
.../pe/DataProcessorPipelineElementManagement.java | 4 +-
.../pe/DataSinkPipelineElementManagement.java | 4 +-
.../pe/InvocablePipelineElementManagement.java | 4 +-
.../http/InvokePipelineElementSubmitter.java | 2 +-
.../http/PipelineElementExtensionRequest.java | 5 +-
.../http/PipelineElementRollbackTest.java | 136 +++++++++++++++++++++
.../runtime/StandaloneEventProcessorRuntime.java | 57 ++++++++-
.../runtime/StandaloneEventSinkRuntime.java | 45 ++++++-
.../runtime/StandalonePipelineElementRuntime.java | 77 +++++++++++-
9 files changed, 313 insertions(+), 21 deletions(-)
diff --git
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataProcessorPipelineElementManagement.java
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataProcessorPipelineElementManagement.java
index 969753f753..7e0af38a5a 100644
---
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataProcessorPipelineElementManagement.java
+++
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataProcessorPipelineElementManagement.java
@@ -73,7 +73,9 @@ public class DataProcessorPipelineElementManagement extends
InvocablePipelineEle
DataProcessorInvocation graph) {
var runtime = getRuntime();
var response = runtime.onRuntimeInvoked(runningInstanceId,
pipelineElement, graph);
- getRunningInstances().add(runningInstanceId, graph, runtime);
+ if (response.isSuccess()) {
+ getRunningInstances().add(runningInstanceId, graph, runtime);
+ }
return response;
}
}
diff --git
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataSinkPipelineElementManagement.java
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataSinkPipelineElementManagement.java
index 6e61831878..b7e59e2956 100644
---
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataSinkPipelineElementManagement.java
+++
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/DataSinkPipelineElementManagement.java
@@ -72,7 +72,9 @@ public class DataSinkPipelineElementManagement extends
InvocablePipelineElementM
DataSinkInvocation graph) {
var runtime = getRuntime();
var response = runtime.onRuntimeInvoked(runningInstanceId,
pipelineElement, graph);
- getRunningInstances().add(runningInstanceId, graph, runtime);
+ if (response.isSuccess()) {
+ getRunningInstances().add(runningInstanceId, graph, runtime);
+ }
return response;
}
}
diff --git
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
index d3125dac3d..4e67c458e9 100644
---
a/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
+++
b/streampipes-extensions-management/src/main/java/org/apache/streampipes/extensions/management/pe/InvocablePipelineElementManagement.java
@@ -122,9 +122,7 @@ public abstract class InvocablePipelineElementManagement<
if (runningInstance != null) {
Response response = runningInstance.onRuntimeDetached(runningInstanceId);
- if (response.isSuccess()) {
- runningInstances.remove(runningInstanceId);
- }
+ runningInstances.remove(runningInstanceId);
return response;
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokePipelineElementSubmitter.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokePipelineElementSubmitter.java
index f8e57cc83f..1ea74c7d56 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokePipelineElementSubmitter.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokePipelineElementSubmitter.java
@@ -72,7 +72,7 @@ public class InvokePipelineElementSubmitter extends
BasePipelineElementSubmitter
List<InvocableStreamPipesEntity> pe) {
return pe
.stream()
- .filter(g -> g.getBelongsTo().equals(elementId))
+ .filter(g -> g.getElementId().equals(elementId))
.findFirst();
}
}
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementExtensionRequest.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementExtensionRequest.java
index 081a07bb7e..1f108cb4be 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementExtensionRequest.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineElementExtensionRequest.java
@@ -71,13 +71,12 @@ public abstract class PipelineElementExtensionRequest {
org.apache.streampipes.model.Response streamPipesResp = JacksonSerializer
.getObjectMapper()
.readValue(resp, org.apache.streampipes.model.Response.class);
- return convert(streamPipesResp, endpointUrl, pipelineElement.getName());
+ return convert(streamPipesResp, pipelineElement.getName());
}
private PipelineElementStatus convert(org.apache.streampipes.model.Response
response,
- String endpointUrl,
String pipelineElementName) {
- return new PipelineElementStatus(endpointUrl, pipelineElementName,
response.isSuccess(),
+ return new PipelineElementStatus(response.getElementId(),
pipelineElementName, response.isSuccess(),
response.getOptionalMessage());
}
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/execution/http/PipelineElementRollbackTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/execution/http/PipelineElementRollbackTest.java
new file mode 100644
index 0000000000..39ded2b08d
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/execution/http/PipelineElementRollbackTest.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.manager.execution.http;
+
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceOperationResult;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineElementStatus;
+import org.apache.streampipes.resource.management.SpResourceManager;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.Mockito.mock;
+
+class PipelineElementRollbackTest {
+
+ @Test
+ void extensionResponseUsesRuntimeElementIdInStatus() {
+ var request = new TestPipelineElementExtensionRequest();
+ var element = makeElement("urn:streampipes:runtime-element", "app-id",
"service-id", "Element");
+
+ var status = request.execute(element, "pipeline-id");
+
+ assertEquals("urn:streampipes:runtime-element", status.getElementId());
+ }
+
+ @Test
+ void rollbackDetachesSuccessfullyStartedElementByRuntimeElementId() {
+ var startedElement = makeElement("urn:streampipes:started", "app-started",
"service-id", "Started");
+ var failedElement = makeElement("urn:streampipes:failed", "app-failed",
"service-id", "Failed");
+ var submitter = new TestInvokePipelineElementSubmitter(
+ Map.of(
+ startedElement.getElementId(), new
PipelineElementStatus(startedElement.getElementId(),
+ startedElement.getName(), true, ""),
+ failedElement.getElementId(), new
PipelineElementStatus(failedElement.getElementId(),
+ failedElement.getName(), false, "startup failed")
+ )
+ );
+
+ var status = submitter.submit(List.of(startedElement, failedElement));
+
+ assertFalse(status.isSuccess());
+ assertEquals(List.of(startedElement), submitter.detachedElements);
+ }
+
+ private static DataSinkInvocation makeElement(String elementId,
+ String belongsTo,
+ String selectedServiceId,
+ String name) {
+ var element = new DataSinkInvocation();
+ element.setElementId(elementId);
+ element.setBelongsTo(belongsTo);
+ element.setSelectedServiceId(selectedServiceId);
+ element.setName(name);
+ return element;
+ }
+
+ private static class TestPipelineElementExtensionRequest extends
PipelineElementExtensionRequest {
+
+ TestPipelineElementExtensionRequest() {
+ super(mock(ExtensionServiceRequestManager.class),
mock(SpResourceManager.class));
+ }
+
+ @Override
+ protected ExtensionServiceOperationResult
performRequest(InvocableStreamPipesEntity pipelineElement,
+ String
pipelineId) throws IOException {
+ var body = """
+
{"elementId":"urn:streampipes:runtime-element","success":true,"optionalMessage":""}
+ """.getBytes(StandardCharsets.UTF_8);
+ return new ExtensionServiceOperationResult(200, body);
+ }
+
+ @Override
+ protected void logError(String endpointUrl,
+ String pipelineElementName,
+ String exceptionMessage) {
+
+ }
+ }
+
+ private static class TestInvokePipelineElementSubmitter extends
InvokePipelineElementSubmitter {
+
+ private final Map<String, PipelineElementStatus> responses;
+ private final List<InvocableStreamPipesEntity> detachedElements;
+
+ TestInvokePipelineElementSubmitter(Map<String, PipelineElementStatus>
responses) {
+ super(makePipeline(), mock(ExtensionServiceRequestManager.class),
mock(SpResourceManager.class));
+ this.responses = responses;
+ this.detachedElements = new ArrayList<>();
+ }
+
+ @Override
+ protected PipelineElementStatus submitElement(InvocableStreamPipesEntity
pipelineElement) {
+ return responses.get(pipelineElement.getElementId());
+ }
+
+ @Override
+ protected PipelineElementStatus performDetach(InvocableStreamPipesEntity
pipelineElement) {
+ detachedElements.add(pipelineElement);
+ return new PipelineElementStatus(pipelineElement.getElementId(),
pipelineElement.getName(), true, "");
+ }
+
+ private static Pipeline makePipeline() {
+ var pipeline = new Pipeline();
+ pipeline.setPipelineId("pipeline-id");
+ pipeline.setName("Pipeline");
+ return pipeline;
+ }
+ }
+}
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
index aebf71046a..f81463fb06 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventProcessorRuntime.java
@@ -48,6 +48,7 @@ public class StandaloneEventProcessorRuntime extends
StandalonePipelineElementRu
private static final Logger LOG =
LoggerFactory.getLogger(StandaloneEventProcessorRuntime.class);
protected SpOutputCollector outputCollector;
+ private boolean pipelineStarted;
public StandaloneEventProcessorRuntime() {
super(new DataProcessorContextGenerator(), new
DataProcessorParameterGenerator());
@@ -96,14 +97,64 @@ public class StandaloneEventProcessorRuntime extends
StandalonePipelineElementRu
protected void beforeStart() {
this.outputCollector = getOutputCollector();
pipelineElement.onPipelineStarted(runtimeParameters, outputCollector,
runtimeContext);
+ pipelineStarted = true;
prepareRuntime();
}
@Override
protected void afterStop() {
- disconnectInputCollectors();
- pipelineElement.onPipelineStopped();
- outputCollector.disconnect();
+ RuntimeException stopException = null;
+ try {
+ disconnectInputCollectors();
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ }
+ try {
+ pipelineElement.onPipelineStopped();
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ } finally {
+ pipelineStarted = false;
+ }
+ try {
+ outputCollector.disconnect();
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ }
+
+ if (stopException != null) {
+ throw stopException;
+ }
+ }
+
+ @Override
+ protected void afterStartFailed() {
+ RuntimeException cleanupException = null;
+ try {
+ super.afterStartFailed();
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ }
+ try {
+ if (pipelineStarted) {
+ pipelineElement.onPipelineStopped();
+ }
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ } finally {
+ pipelineStarted = false;
+ }
+ try {
+ if (outputCollector != null) {
+ outputCollector.disconnect();
+ }
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ }
+
+ if (cleanupException != null) {
+ throw cleanupException;
+ }
}
}
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java
index 859468a761..c8046673a5 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandaloneEventSinkRuntime.java
@@ -45,6 +45,7 @@ public class StandaloneEventSinkRuntime extends
StandalonePipelineElementRuntime
IDataSinkParameters> implements IDataSinkRuntime, RawDataProcessor {
private static final Logger LOG =
LoggerFactory.getLogger(StandaloneEventSinkRuntime.class);
+ private boolean pipelineStarted;
public StandaloneEventSinkRuntime() {
super(new DataSinkContextGenerator(), new DataSinkParameterGenerator());
@@ -83,14 +84,52 @@ public class StandaloneEventSinkRuntime extends
StandalonePipelineElementRuntime
@Override
protected void beforeStart() {
pipelineElement.onPipelineStarted(runtimeParameters, runtimeContext);
+ pipelineStarted = true;
inputCollectors.forEach(is -> is.registerConsumer(instanceId, this));
prepareRuntime();
}
@Override
protected void afterStop() {
- inputCollectors.forEach(is -> is.unregisterConsumer(instanceId));
- pipelineElement.onPipelineStopped();
- postDiscard();
+ RuntimeException stopException = null;
+ try {
+ pipelineElement.onPipelineStopped();
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ } finally {
+ pipelineStarted = false;
+ }
+ try {
+ postDiscard();
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ }
+
+ if (stopException != null) {
+ throw stopException;
+ }
+ }
+
+ @Override
+ protected void afterStartFailed() {
+ RuntimeException cleanupException = null;
+ try {
+ super.afterStartFailed();
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ }
+ try {
+ if (pipelineStarted) {
+ pipelineElement.onPipelineStopped();
+ }
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ } finally {
+ pipelineStarted = false;
+ }
+
+ if (cleanupException != null) {
+ throw cleanupException;
+ }
}
}
diff --git
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java
index e906161004..547a4e7de9 100644
---
a/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java
+++
b/streampipes-wrapper-standalone/src/main/java/org/apache/streampipes/wrapper/standalone/runtime/StandalonePipelineElementRuntime.java
@@ -75,18 +75,44 @@ public abstract class StandalonePipelineElementRuntime<
this.runtimeContext = runtimeContext;
this.instanceId = pipelineElementInvocation.getElementId();
this.inputCollectors =
getInputCollectors(pipelineElementInvocation.getInputStreams());
- this.beforeStart();
+ try {
+ this.beforeStart();
+ } catch (RuntimeException e) {
+ try {
+ this.afterStartFailed();
+ } catch (RuntimeException cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
+ throw e;
+ }
}
@Override
public void stopRuntime() {
- this.inputCollectors.forEach(is -> is.unregisterConsumer(instanceId));
- resetCounter(instanceId);
- afterStop();
+ RuntimeException stopException = null;
+ try {
+ unregisterInputCollectors();
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ }
+ try {
+ afterStop();
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ }
+ try {
+ removeMonitoring(instanceId);
+ } catch (RuntimeException e) {
+ stopException = collectCleanupException(stopException, e);
+ }
+
+ if (stopException != null) {
+ throw stopException;
+ }
}
- protected void resetCounter(String resourceId) throws SpRuntimeException {
- monitoringManager.resetCounter(resourceId);
+ protected void removeMonitoring(String resourceId) throws SpRuntimeException
{
+ monitoringManager.remove(resourceId);
}
protected List<SpInputCollector> getInputCollectors(List<SpDataStream>
inputStreams) throws SpRuntimeException {
@@ -115,6 +141,45 @@ public abstract class StandalonePipelineElementRuntime<
this.inputCollectors.forEach(is -> is.registerConsumer(instanceId, this));
}
+ protected void unregisterInputCollectors() {
+ if (this.inputCollectors != null) {
+ this.inputCollectors.forEach(is -> is.unregisterConsumer(instanceId));
+ }
+ }
+
+ protected void afterStartFailed() {
+ RuntimeException cleanupException = null;
+ try {
+ unregisterInputCollectors();
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ }
+ try {
+ disconnectInputCollectors();
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ }
+ try {
+ removeMonitoring(instanceId);
+ } catch (RuntimeException e) {
+ cleanupException = collectCleanupException(cleanupException, e);
+ }
+
+ if (cleanupException != null) {
+ throw cleanupException;
+ }
+ }
+
+ protected RuntimeException collectCleanupException(RuntimeException
cleanupException,
+ RuntimeException
nextException) {
+ if (cleanupException == null) {
+ return nextException;
+ } else {
+ cleanupException.addSuppressed(nextException);
+ return cleanupException;
+ }
+ }
+
protected abstract void beforeStart();
protected abstract void afterStop();
