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 9c3781c4ee fix: keep output schema fallback for unconfigured elements
(#5192)
9c3781c4ee is described below
commit 9c3781c4eeb2af8c2aed012331d518abd1125041
Author: Philipp Zehnder <[email protected]>
AuthorDate: Tue Nov 3 11:28:26 2026 +0100
fix: keep output schema fallback for unconfigured elements (#5192)
---
.../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() {
