This is an automated email from the ASF dual-hosted git repository. tenthe pushed a commit to branch fix-cypress-pipeline-dataset-schema-update in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit d2c5349f96b701e45618eebd8d6ea92101bcfa22 Author: Philipp Zehnder <[email protected]> AuthorDate: Tue Nov 3 09:22:22 2026 +0100 fix: keep output schema fallback for unconfigured elements --- .../output/CustomTransformOutputSchemaGenerator.java | 20 +++++++++++++------- .../CustomTransformOutputSchemaGeneratorTest.java | 14 ++++++-------- 2 files changed, 19 insertions(+), 15 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 cc0a04812c..dbd6b1a28a 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 @@ -36,12 +36,16 @@ import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; import com.google.gson.JsonSyntaxException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.Set; public class CustomTransformOutputSchemaGenerator extends OutputSchemaGenerator<CustomTransformOutputStrategy> { + private static final Logger LOG = LoggerFactory.getLogger(CustomTransformOutputSchemaGenerator.class); + private final ISpServiceDiscovery serviceDiscovery; private final DataProcessorInvocation dataProcessorInvocation; private final ExtensionServiceRequestManager requestManager; @@ -117,23 +121,25 @@ public class CustomTransformOutputSchemaGenerator extends OutputSchemaGenerator< } if (!response.isSuccess()) { - var cause = new IOException("Could not compute output schema, status code: " + response.statusCode()); if (isUnavailableStatus(response.statusCode())) { + var cause = new IOException("Could not compute output schema, status code: " + response.statusCode()); throw new OutputSchemaUnavailableException(dataProcessorInvocation.getAppId(), cause); } - throw new IllegalStateException(cause); + // The extension is reachable but cannot compute a schema yet, e.g. while the element is not + // fully configured in the pipeline editor. Keep the pipeline editable with an empty schema. + LOG.warn("Could not compute output schema for {}, status code: {}", + dataProcessorInvocation.getAppId(), response.statusCode()); + return new EventSchema(); } try { var eventSchema = JacksonSerializer .getObjectMapper() .readValue(response.responseBody(), EventSchema.class); - if (eventSchema == null) { - throw new IllegalStateException("The extension returned no output schema"); - } - return eventSchema; + return eventSchema != null ? eventSchema : new EventSchema(); } catch (IOException | JsonSyntaxException e) { - throw new IllegalStateException("The extension returned an invalid output schema", e); + LOG.warn("The extension {} returned an invalid output schema", dataProcessorInvocation.getAppId(), e); + return new EventSchema(); } } 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 0c256a82c7..b69394e06b 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 @@ -91,7 +91,7 @@ class CustomTransformOutputSchemaGeneratorTest { } @Test - void throwsWhenTheExtensionReturnsAnErrorResponse() throws IOException { + void returnsEmptySchemaWhenTheExtensionReturnsAnErrorResponse() throws IOException { var invocation = invocation(); var serviceDiscovery = mock(ISpServiceDiscovery.class); var requestManager = mock(ExtensionServiceRequestManager.class); @@ -104,10 +104,9 @@ class CustomTransformOutputSchemaGeneratorTest { requestManager ); - var exception = assertThrows(IllegalStateException.class, - () -> generator.buildFromOneStream(null)); + var schema = generator.buildFromOneStream(null).k; - assertTrue(exception.getMessage().contains("status code: 500")); + assertTrue(schema.getEventProperties().isEmpty()); } @Test @@ -131,7 +130,7 @@ class CustomTransformOutputSchemaGeneratorTest { } @Test - void throwsWhenTheExtensionReturnsAnInvalidSchema() throws IOException { + void returnsEmptySchemaWhenTheExtensionReturnsAnInvalidSchema() throws IOException { var invocation = invocation(); var serviceDiscovery = mock(ISpServiceDiscovery.class); var requestManager = mock(ExtensionServiceRequestManager.class); @@ -145,10 +144,9 @@ class CustomTransformOutputSchemaGeneratorTest { requestManager ); - var exception = assertThrows(IllegalStateException.class, - () -> generator.buildFromOneStream(null)); + var schema = generator.buildFromOneStream(null).k; - assertTrue(exception.getMessage().contains("invalid output schema")); + assertTrue(schema.getEventProperties().isEmpty()); } private DataProcessorInvocation invocation() {
