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 6543af8fad refactor: Change data stream update event to only publish
schema changes (#5163)
6543af8fad is described below
commit 6543af8fad6331dd70c3820f4072083e4df1df92
Author: Sven Oehler <[email protected]>
AuthorDate: Fri Oct 30 12:25:35 2026 +0100
refactor: Change data stream update event to only publish schema changes
(#5163)
---
.../management/AdapterUpdateManagement.java | 23 +---
.../management/AdapterSchemaUpdateTest.java | 94 ++++++++++++++++
.../management/AdapterUpdateAuditTest.java | 8 +-
...vent.java => DataStreamSchemaUpdatedEvent.java} | 4 +-
.../update/DataStreamUpdateManagement.java | 14 ++-
.../pipeline/update/EventSchemaChangeDetector.java | 84 ++++++++++++++
.../update/DataStreamUpdateManagementTest.java | 115 +++++++++++++++++++
.../update/EventSchemaChangeDetectorTest.java | 122 +++++++++++++++++++++
.../rest/impl/connect/AdapterResource.java | 13 +--
.../rest/impl/connect/CompactAdapterResource.java | 7 +-
.../rest/impl/pe/DataStreamResource.java | 10 +-
.../core/PipelineManagementConfiguration.java | 11 ++
.../core/storage/StorageApiConfigurationTest.java | 5 +
13 files changed, 460 insertions(+), 50 deletions(-)
diff --git
a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java
index e5d8eb53be..8bdf0b62fd 100644
---
a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java
+++
b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java
@@ -22,9 +22,7 @@ import org.apache.streampipes.audit.api.AuditOutcome;
import org.apache.streampipes.audit.events.adapter.AdapterAuditRecorder;
import org.apache.streampipes.audit.events.adapter.AdapterEditedDetails;
import org.apache.streampipes.commons.exceptions.connect.AdapterException;
-import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import
org.apache.streampipes.manager.pipeline.update.DataStreamUpdateManagement;
-import org.apache.streampipes.manager.pipeline.update.DataStreamUpdatedEvent;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
import org.apache.streampipes.model.connect.adapter.PipelineUpdateInfo;
@@ -32,8 +30,6 @@ import
org.apache.streampipes.resource.management.AdapterResourceManager;
import org.apache.streampipes.resource.management.DataStreamResourceManager;
import org.apache.streampipes.resource.management.SpResourceManager;
-import org.springframework.context.ApplicationEventPublisher;
-
import java.util.List;
public class AdapterUpdateManagement {
@@ -44,21 +40,15 @@ public class AdapterUpdateManagement {
private final AdapterResourceManager adapterResourceManager;
private final DataStreamResourceManager dataStreamResourceManager;
private final DataStreamUpdateManagement dataStreamUpdateManagement;
- private final ApplicationEventPublisher eventPublisher;
public AdapterUpdateManagement(AdapterMasterManagement
adapterMasterManagement,
- ExtensionServiceRequestManager requestManager,
SpResourceManager resourceManager,
- ApplicationEventPublisher eventPublisher) {
+ DataStreamUpdateManagement
dataStreamUpdateManagement) {
this.adapterAudit = new
AdapterAuditRecorder(resourceManager.getAuditService());
this.adapterMasterManagement = adapterMasterManagement;
this.adapterResourceManager = resourceManager.manageAdapters();
this.dataStreamResourceManager = resourceManager.manageDataStreams();
- this.dataStreamUpdateManagement = new DataStreamUpdateManagement(
- requestManager,
- resourceManager
- );
- this.eventPublisher = eventPublisher;
+ this.dataStreamUpdateManagement = dataStreamUpdateManagement;
}
public void updateAdapter(AdapterDescription ad)
@@ -84,7 +74,6 @@ public class AdapterUpdateManagement {
// update data source
var updatedDataStream = this.updateDataSource(ad);
dataStreamUpdateManagement.updateDataStream(updatedDataStream);
- publishEvent(new DataStreamUpdatedEvent(updatedDataStream));
if (shouldRestart) {
this.adapterMasterManagement.startAdapter(ad.getElementId(), actor);
@@ -104,7 +93,7 @@ public class AdapterUpdateManagement {
// get data source
SpDataStream dataStream =
this.dataStreamResourceManager.find(ad.getCorrespondingDataStreamElementId());
- return SourcesManagement.updateDataStream(ad, dataStream);
+ return SourcesManagement.updateDataStream(ad, new
SpDataStream(dataStream));
}
private SpDataStream toDataStreamUpdate(AdapterDescription
adapterDescription) {
@@ -115,10 +104,4 @@ public class AdapterUpdateManagement {
dataStream.setEventSchema(adapterDescription.getEventSchema());
return dataStream;
}
-
- private void publishEvent(Object event) {
- if (eventPublisher != null) {
- eventPublisher.publishEvent(event);
- }
- }
}
diff --git
a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterSchemaUpdateTest.java
b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterSchemaUpdateTest.java
new file mode 100644
index 0000000000..347affff1f
--- /dev/null
+++
b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterSchemaUpdateTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.connect.management.management;
+
+import org.apache.streampipes.audit.api.AuditService;
+import
org.apache.streampipes.manager.pipeline.update.DataStreamSchemaUpdatedEvent;
+import
org.apache.streampipes.manager.pipeline.update.DataStreamUpdateManagement;
+import
org.apache.streampipes.manager.pipeline.update.PipelineUpdateCoordinator;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.resource.management.AdapterResourceManager;
+import org.apache.streampipes.resource.management.DataStreamResourceManager;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.storage.api.connect.IAdapterStorage;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.context.ApplicationEventPublisher;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+class AdapterSchemaUpdateTest {
+
+ @Test
+ void publishesSchemaEventWhenAdapterFieldAdded() throws Exception {
+ verifyUpdate(true);
+ }
+
+ @Test
+ void doesNotPublishSchemaEventWhenAdapterRenamed() throws Exception {
+ verifyUpdate(false);
+ }
+
+ private void verifyUpdate(boolean schemaChanged) throws Exception {
+ var resources = mock(SpResourceManager.class);
+ var adapters = mock(AdapterResourceManager.class);
+ var streams = mock(DataStreamResourceManager.class);
+ var storage = mock(IAdapterStorage.class);
+ var publisher = mock(ApplicationEventPublisher.class);
+ when(resources.getAuditService()).thenReturn(mock(AuditService.class));
+ when(resources.manageAdapters()).thenReturn(adapters);
+ when(resources.manageDataStreams()).thenReturn(streams);
+ when(adapters.getDb()).thenReturn(storage);
+ when(storage.getElementById("adapter-1")).thenReturn(new
AdapterDescription());
+ var previous = new SpDataStream();
+ previous.setElementId("stream-1");
+ when(streams.find("stream-1")).thenReturn(previous);
+ var updated = new AdapterDescription();
+ updated.setElementId("adapter-1");
+ updated.setCorrespondingDataStreamElementId("stream-1");
+ updated.setName("Renamed adapter");
+ var adapterStream = new SpDataStream(previous);
+ if (schemaChanged) {
+ adapterStream.getEventSchema().addEventProperty(new
EventPropertyPrimitive("xs:double", "value", null, null));
+ }
+ updated.setDataStream(adapterStream);
+
+ try (var ignored = mockConstruction(PipelineUpdateCoordinator.class)) {
+ new AdapterUpdateManagement(mock(AdapterMasterManagement.class),
resources,
+ new DataStreamUpdateManagement(null, resources, publisher))
+ .updateAdapter(updated, "user-1");
+
+ verify(streams).update(any(SpDataStream.class));
+ assertEquals(0, previous.getEventSchema().getEventProperties().size());
+ if (schemaChanged) {
+
verify(publisher).publishEvent(any(DataStreamSchemaUpdatedEvent.class));
+ } else {
+ verifyNoInteractions(publisher);
+ }
+ }
+ }
+}
diff --git
a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterUpdateAuditTest.java
b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterUpdateAuditTest.java
index fde3170441..dc254c7233 100644
---
a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterUpdateAuditTest.java
+++
b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterUpdateAuditTest.java
@@ -40,7 +40,6 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
@@ -84,6 +83,8 @@ class AdapterUpdateAuditTest {
before.setName("old");
var after = new AdapterDescription();
after.setElementId("adapter-1");
+ after.setCorrespondingDataStreamElementId("stream-1");
+ when(streams.find("stream-1")).thenReturn(new SpDataStream());
after.setName(changed ? "new" : "old");
after.setRunning(true);
when(storage.getElementById("adapter-1")).thenReturn(before);
@@ -93,10 +94,9 @@ class AdapterUpdateAuditTest {
if (failRestart) {
doThrow(new
AdapterException("private")).when(master).startAdapter("adapter-1", "user-1");
}
- try (var updates = mockConstruction(DataStreamUpdateManagement.class);
- var sources = mockStatic(SourcesManagement.class)) {
+ try (var sources = mockStatic(SourcesManagement.class)) {
sources.when(() -> SourcesManagement.updateDataStream(any(),
any())).thenReturn(new SpDataStream());
- var manager = new AdapterUpdateManagement(master, null, resources, null);
+ var manager = new AdapterUpdateManagement(master, resources,
mock(DataStreamUpdateManagement.class));
if (failPersistence || failRestart) {
assertThrows(AdapterException.class, () ->
manager.updateAdapter(after, "user-1"));
} else {
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdatedEvent.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamSchemaUpdatedEvent.java
similarity index 88%
rename from
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdatedEvent.java
rename to
streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamSchemaUpdatedEvent.java
index 76bb4622c6..abdf023a75 100644
---
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdatedEvent.java
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/DataStreamSchemaUpdatedEvent.java
@@ -18,7 +18,5 @@
package org.apache.streampipes.manager.pipeline.update;
-import org.apache.streampipes.model.SpDataStream;
-
-public record DataStreamUpdatedEvent(SpDataStream dataStream) {
+public record DataStreamSchemaUpdatedEvent(String elementId, String name) {
}
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 9fe004577c..5de6578114 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
@@ -24,22 +24,34 @@ import
org.apache.streampipes.model.connect.adapter.PipelineUpdateInfo;
import org.apache.streampipes.resource.management.DataStreamResourceManager;
import org.apache.streampipes.resource.management.SpResourceManager;
+import org.springframework.context.ApplicationEventPublisher;
+
import java.util.List;
+import java.util.Objects;
public class DataStreamUpdateManagement {
private final DataStreamResourceManager dataStreamResourceManager;
private final PipelineUpdateCoordinator pipelineUpdateCoordinator;
+ private final ApplicationEventPublisher eventPublisher;
public DataStreamUpdateManagement(ExtensionServiceRequestManager
requestManager,
- SpResourceManager resourceManager) {
+ SpResourceManager resourceManager,
+ ApplicationEventPublisher eventPublisher) {
+ this.eventPublisher = Objects.requireNonNull(eventPublisher);
this.pipelineUpdateCoordinator = new
PipelineUpdateCoordinator(requestManager, resourceManager);
this.dataStreamResourceManager = resourceManager.manageDataStreams();
}
public void updateDataStream(SpDataStream dataStream) {
+ var previous = dataStreamResourceManager.find(dataStream.getElementId());
+ boolean schemaChanged = EventSchemaChangeDetector.hasChanged(
+ previous == null ? null : previous.getEventSchema(),
dataStream.getEventSchema());
dataStreamResourceManager.update(dataStream);
pipelineUpdateCoordinator.updatePipelines(dataStream);
+ if (schemaChanged) {
+ eventPublisher.publishEvent(new
DataStreamSchemaUpdatedEvent(dataStream.getElementId(), dataStream.getName()));
+ }
}
public List<PipelineUpdateInfo> checkPipelineMigrations(SpDataStream
dataStream) {
diff --git
a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/EventSchemaChangeDetector.java
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/EventSchemaChangeDetector.java
new file mode 100644
index 0000000000..d300b2b396
--- /dev/null
+++
b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/EventSchemaChangeDetector.java
@@ -0,0 +1,84 @@
+/*
+ * 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.pipeline.update;
+
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyList;
+import org.apache.streampipes.model.schema.EventPropertyNested;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+final class EventSchemaChangeDetector {
+
+ private EventSchemaChangeDetector() {
+ }
+
+ static boolean hasChanged(EventSchema previous, EventSchema updated) {
+ return !sameProperties(previous == null ? null :
previous.getEventProperties(),
+ updated == null ? null : updated.getEventProperties());
+ }
+
+ private static boolean sameProperties(List<EventProperty> previous,
List<EventProperty> updated) {
+ var previousProperties = previous == null ? List.<EventProperty>of() :
previous;
+ var unmatched = new ArrayList<>(updated == null ? List.<EventProperty>of()
: updated);
+ if (previousProperties.size() != unmatched.size()) {
+ return false;
+ }
+ // Match fields independently of order, including fields within nested
objects.
+ for (var property : previousProperties) {
+ boolean matched = false;
+ for (int i = 0; i < unmatched.size(); i++) {
+ if (sameProperty(property, unmatched.get(i))) {
+ unmatched.remove(i);
+ matched = true;
+ break;
+ }
+ }
+ if (!matched) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean sameProperty(EventProperty previous, EventProperty
updated) {
+ if (previous == null || updated == null) {
+ return previous == updated;
+ }
+ if (previous.getClass() != updated.getClass()
+ || !Objects.equals(previous.getRuntimeName(), updated.getRuntimeName())
+ || !Objects.equals(previous.getPropertyScope(),
updated.getPropertyScope())) {
+ return false;
+ }
+ if (previous instanceof EventPropertyPrimitive primitive) {
+ return Objects.equals(primitive.getRuntimeType(),
((EventPropertyPrimitive) updated).getRuntimeType());
+ }
+ if (previous instanceof EventPropertyNested nested) {
+ return sameProperties(nested.getEventProperties(),
((EventPropertyNested) updated).getEventProperties());
+ }
+ if (previous instanceof EventPropertyList list) {
+ return sameProperty(list.getEventProperty(), ((EventPropertyList)
updated).getEventProperty());
+ }
+ return true;
+ }
+}
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
new file mode 100644
index 0000000000..97e16c7f0e
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/DataStreamUpdateManagementTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.pipeline.update;
+
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.resource.management.DataStreamResourceManager;
+import org.apache.streampipes.resource.management.SpResourceManager;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.context.ApplicationEventPublisher;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+class DataStreamUpdateManagementTest {
+
+ @Test
+ void publishesSchemaEventAfterSuccessfulUpdate() {
+ var previous = stream();
+ var updated = new SpDataStream(previous);
+ updated.getEventSchema().addEventProperty(new
EventPropertyPrimitive("xs:string", "new-field", null, null));
+ 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))
{
+ var management = new DataStreamUpdateManagement(null, resources,
publisher);
+ management.updateDataStream(updated);
+
+ var order = inOrder(streams, coordinators.constructed().getFirst(),
publisher);
+ order.verify(streams).find("stream-1");
+ order.verify(streams).update(updated);
+
order.verify(coordinators.constructed().getFirst()).updatePipelines(updated);
+ order.verify(publisher).publishEvent(new
DataStreamSchemaUpdatedEvent(updated.getElementId(), updated.getName()));
+ }
+ }
+
+ @Test
+ void persistsStreamRenameWithoutSchemaEvent() {
+ var previous = stream();
+ var updated = new SpDataStream(previous);
+ updated.setName("Renamed stream");
+ updated.setDescription("New description");
+ 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);
+ verify(coordinators.constructed().getFirst()).updatePipelines(updated);
+ verifyNoInteractions(publisher);
+ }
+ }
+
+ @Test
+ void doesNotPublishWhenPersistenceFails() {
+ var updated = stream();
+ updated.getEventSchema().addEventProperty(new
EventPropertyPrimitive("xs:string", "new-field", null, null));
+ 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(stream());
+ doThrow(new IllegalStateException("Persistence
failed")).when(streams).update(updated);
+
+ try (var coordinators = mockConstruction(PipelineUpdateCoordinator.class))
{
+ var management = new DataStreamUpdateManagement(null, resources,
publisher);
+ assertThrows(IllegalStateException.class, () ->
management.updateDataStream(updated));
+
+ verifyNoInteractions(publisher, coordinators.constructed().getFirst());
+ }
+ }
+
+ @Test
+ void requiresEventPublisher() {
+ assertThrows(NullPointerException.class,
+ () -> new DataStreamUpdateManagement(null,
mock(SpResourceManager.class), null));
+ }
+
+ private SpDataStream stream() {
+ var stream = new SpDataStream();
+ stream.setElementId("stream-1");
+ stream.setName("Stream");
+ return stream;
+ }
+}
diff --git
a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/EventSchemaChangeDetectorTest.java
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/EventSchemaChangeDetectorTest.java
new file mode 100644
index 0000000000..81d0b6086f
--- /dev/null
+++
b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/update/EventSchemaChangeDetectorTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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.pipeline.update;
+
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventPropertyList;
+import org.apache.streampipes.model.schema.EventPropertyNested;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.net.URI;
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class EventSchemaChangeDetectorTest {
+
+ @ParameterizedTest
+ @MethodSource("schemaChanges")
+ void detectsFieldChanges(EventSchema previous, EventSchema updated) {
+ assertTrue(EventSchemaChangeDetector.hasChanged(previous, updated));
+ assertTrue(EventSchemaChangeDetector.hasChanged(updated, previous));
+ }
+
+ private static Stream<Arguments> schemaChanges() {
+ var original = schema(field("value", "xs:string", "measurement"));
+ return Stream.of(
+ Arguments.of(original, schema()),
+ Arguments.of(original, schema(field("value", "xs:string",
"measurement"), field("extra"))),
+ Arguments.of(original, schema(field("renamed", "xs:string",
"measurement"))),
+ Arguments.of(original, schema(field("value", "xs:double",
"measurement"))),
+ Arguments.of(original, schema(field("value", "xs:string",
"dimension"))),
+ Arguments.of(original, schema(field("value", "xs:string", null))),
+ Arguments.of(schema(new EventPropertyNested("object",
List.of(field("child")))),
+ schema(new EventPropertyNested("object",
List.of(field("renamed"))))),
+ Arguments.of(schema(new EventPropertyNested("object",
List.of(field("child")))),
+ schema(new EventPropertyNested("object", List.of()))),
+ Arguments.of(schema(new EventPropertyList("values", field("item"))),
+ schema(new EventPropertyList("values", field("item", "xs:double",
null)))),
+ Arguments.of(schema(new EventPropertyList("values", new
EventPropertyNested("item", List.of(field("child"))))),
+ schema(new EventPropertyList("values", new
EventPropertyNested("item", List.of(field("new")))))),
+ Arguments.of(schema(field("value")), schema(new
EventPropertyNested("value", List.of()))),
+ Arguments.of(schema(field("value")), schema(new
EventPropertyList("value", field("item")))),
+ Arguments.of(null, schema(field("value")))
+ );
+ }
+
+ @Test
+ void ignoresFieldOrderAtEveryLevel() {
+ var previous = schema(field("a"), new EventPropertyNested("object",
List.of(field("b"), field("c"))),
+ new EventPropertyList("list", new EventPropertyNested("item",
List.of(field("d"), field("e")))));
+ var updated = schema(
+ new EventPropertyList("list", new EventPropertyNested("item",
List.of(field("e"), field("d")))),
+ new EventPropertyNested("object", List.of(field("c"), field("b"))),
field("a"));
+
+ assertFalse(EventSchemaChangeDetector.hasChanged(previous, updated));
+ }
+
+ @Test
+ void ignoresDescriptiveAndGeneratedMetadata() {
+ var previous = field("value");
+ var updated = field("value");
+ updated.setLabel("New label");
+ updated.setDescription("New description");
+ updated.setElementId("new-id");
+ updated.setRuntimeId("new-runtime-id");
+ updated.setSemanticType("new-semantic-type");
+ updated.setMeasurementUnit(URI.create("urn:unit:new"));
+
+ assertFalse(EventSchemaChangeDetector.hasChanged(schema(previous),
schema(updated)));
+ }
+
+ @Test
+ void treatsMissingAndEmptySchemasAsHavingNoFields() {
+ assertFalse(EventSchemaChangeDetector.hasChanged(null, schema()));
+ var missingFields = schema();
+ missingFields.setEventProperties(null);
+ assertFalse(EventSchemaChangeDetector.hasChanged(missingFields, schema()));
+ }
+
+ @Test
+ void detectsDuplicateFieldRemoval() {
+ assertTrue(EventSchemaChangeDetector.hasChanged(schema(field("value"),
field("value")), schema(field("value"))));
+ }
+
+ private static EventSchema schema(EventProperty... properties) {
+ return new EventSchema(List.of(properties));
+ }
+
+ private static EventPropertyPrimitive field(String name) {
+ return field(name, "xs:string", null);
+ }
+
+ private static EventPropertyPrimitive field(String name, String type, String
scope) {
+ var property = new EventPropertyPrimitive(type, name, null, null);
+ property.setPropertyScope(scope);
+ return property;
+ }
+}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
index cbacb0482e..2245858c4c 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java
@@ -28,6 +28,7 @@ import
org.apache.streampipes.connect.management.management.WorkerRestClient;
import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.pipeline.PipelineManager;
import org.apache.streampipes.manager.pipeline.update.DataStreamDeletedEvent;
+import
org.apache.streampipes.manager.pipeline.update.DataStreamUpdateManagement;
import org.apache.streampipes.model.client.user.DefaultRole;
import org.apache.streampipes.model.client.user.Permission;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
@@ -86,18 +87,11 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
private final SpResourceManager resourceManager;
private final IAdapterStorage adapterDescriptionStorage;
- public AdapterResource(WorkerRestClient workerRestClient,
- ExtensionServiceRequestManager requestManager,
- SpResourceManager resourceManager,
- @Qualifier("adapterDescriptionStorage")
IAdapterStorage adapterDescriptionStorage,
- IExtensionsServiceStorage extensionsServiceStorage) {
- this(workerRestClient, requestManager, null, resourceManager,
adapterDescriptionStorage, extensionsServiceStorage);
- }
-
@Autowired
public AdapterResource(WorkerRestClient workerRestClient,
ExtensionServiceRequestManager requestManager,
ApplicationEventPublisher eventPublisher,
+ DataStreamUpdateManagement dataStreamUpdateManagement,
SpResourceManager resourceManager,
@Qualifier("adapterDescriptionStorage")
IAdapterStorage adapterDescriptionStorage,
IExtensionsServiceStorage extensionsServiceStorage) {
@@ -118,9 +112,8 @@ public class AdapterResource extends
AbstractAdapterResource<AdapterMasterManage
);
this.adapterUpdateManagement = new AdapterUpdateManagement(
managementService,
- requestManager,
resourceManager,
- eventPublisher
+ dataStreamUpdateManagement
);
}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java
index 9ad177a823..4ac3a71d1b 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/CompactAdapterResource.java
@@ -32,6 +32,7 @@ import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestMana
import
org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
import org.apache.streampipes.manager.pipeline.PipelineManager;
import
org.apache.streampipes.manager.pipeline.compact.CompactPipelineManagement;
+import
org.apache.streampipes.manager.pipeline.update.DataStreamUpdateManagement;
import org.apache.streampipes.model.client.user.DefaultPrivilege;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
import org.apache.streampipes.model.connect.adapter.compact.CompactAdapter;
@@ -49,7 +50,6 @@ import
org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.ApplicationEventPublisher;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
@@ -76,7 +76,7 @@ public class CompactAdapterResource extends
AbstractAdapterResource<AdapterMaste
public CompactAdapterResource(WorkerRestClient workerRestClient,
ExtensionServiceRequestManager requestManager,
- ApplicationEventPublisher eventPublisher,
+ DataStreamUpdateManagement
dataStreamUpdateManagement,
SpResourceManager resourceManager,
@Qualifier("adapterDescriptionStorage")
IAdapterStorage adapterDescriptionStorage,
IExtensionsServiceStorage
extensionsServiceStorage,
@@ -108,9 +108,8 @@ public class CompactAdapterResource extends
AbstractAdapterResource<AdapterMaste
);
this.adapterUpdateManagement = new AdapterUpdateManagement(
managementService,
- requestManager,
resourceManager,
- eventPublisher
+ dataStreamUpdateManagement
);
}
diff --git
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
index f7063bff4e..947df373e4 100644
---
a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
+++
b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/pe/DataStreamResource.java
@@ -18,10 +18,8 @@
package org.apache.streampipes.rest.impl.pe;
-import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.pipeline.update.DataStreamDeletedEvent;
import
org.apache.streampipes.manager.pipeline.update.DataStreamUpdateManagement;
-import org.apache.streampipes.manager.pipeline.update.DataStreamUpdatedEvent;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.connect.adapter.PipelineUpdateInfo;
import org.apache.streampipes.model.message.Message;
@@ -57,14 +55,11 @@ public class DataStreamResource extends
AbstractAuthGuardedRestResource {
private final ApplicationEventPublisher eventPublisher;
private final DataStreamResourceManager dataStreamResourceManager;
- public DataStreamResource(ExtensionServiceRequestManager requestManager,
+ public DataStreamResource(DataStreamUpdateManagement
dataStreamUpdateManagement,
ApplicationEventPublisher eventPublisher,
SpResourceManager resourceManager) {
this.dataStreamResourceManager = resourceManager.manageDataStreams();
- this.dataStreamUpdateManagement = new DataStreamUpdateManagement(
- requestManager,
- resourceManager
- );
+ this.dataStreamUpdateManagement = dataStreamUpdateManagement;
this.eventPublisher = eventPublisher;
}
@@ -112,7 +107,6 @@ public class DataStreamResource extends
AbstractAuthGuardedRestResource {
"Element ID in path variable does not match element ID in request
body");
} else {
dataStreamUpdateManagement.updateDataStream(updatedElement);
- publishEvent(new DataStreamUpdatedEvent(updatedElement));
return
constructSuccessMessage(NotificationType.STORAGE_SUCCESS.uiNotification());
}
}
diff --git
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PipelineManagementConfiguration.java
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PipelineManagementConfiguration.java
index 77e23708b8..f75163b1f2 100644
---
a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PipelineManagementConfiguration.java
+++
b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PipelineManagementConfiguration.java
@@ -18,9 +18,12 @@
package org.apache.streampipes.service.core;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.extensions.AvailableExtensionsProvider;
+import
org.apache.streampipes.manager.pipeline.update.DataStreamUpdateManagement;
import org.apache.streampipes.manager.setup.InstallationConfiguration;
import org.apache.streampipes.resource.management.ResourceDeletionManager;
+import org.apache.streampipes.resource.management.SpResourceManager;
import org.apache.streampipes.storage.api.connect.IAdapterStorage;
import
org.apache.streampipes.storage.api.pipeline.ICompactPipelineTemplateStorage;
import
org.apache.streampipes.storage.api.pipeline.IPipelineElementDescriptionStorage;
@@ -30,12 +33,20 @@ import
org.apache.streampipes.svcdiscovery.SpServiceDiscoveryCore;
import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class PipelineManagementConfiguration {
+ @Bean
+ public DataStreamUpdateManagement
dataStreamUpdateManagement(ExtensionServiceRequestManager requestManager,
+
SpResourceManager resourceManager,
+
ApplicationEventPublisher eventPublisher) {
+ return new DataStreamUpdateManagement(requestManager, resourceManager,
eventPublisher);
+ }
+
@Bean
public ResourceDeletionManager resourceDeletionManager(IGenericStorage
genericStorage) {
return new ResourceDeletionManager(genericStorage);
diff --git
a/streampipes-service-core/src/test/java/org/apache/streampipes/service/core/storage/StorageApiConfigurationTest.java
b/streampipes-service-core/src/test/java/org/apache/streampipes/service/core/storage/StorageApiConfigurationTest.java
index dc92169e97..17007cd8e0 100644
---
a/streampipes-service-core/src/test/java/org/apache/streampipes/service/core/storage/StorageApiConfigurationTest.java
+++
b/streampipes-service-core/src/test/java/org/apache/streampipes/service/core/storage/StorageApiConfigurationTest.java
@@ -17,9 +17,11 @@
*/
package org.apache.streampipes.service.core.storage;
+import
org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
import org.apache.streampipes.manager.extensions.AvailableExtensionsProvider;
import org.apache.streampipes.manager.setup.InstallationConfiguration;
import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.resource.management.SpResourceManager;
import org.apache.streampipes.service.core.PipelineManagementConfiguration;
import org.apache.streampipes.storage.api.connect.IAdapterStorage;
import org.apache.streampipes.storage.api.function.IFunctionStateStorage;
@@ -81,6 +83,7 @@ import static
org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockStatic;
import static org.mockito.Mockito.when;
@@ -106,6 +109,8 @@ class StorageApiConfigurationTest {
try (var clients = mockStatic(Utils.class);
var context = new AnnotationConfigApplicationContext()) {
context.registerBean(CacheManager.class, () -> cacheManager);
+ context.registerBean(ExtensionServiceRequestManager.class, () ->
mock(ExtensionServiceRequestManager.class));
+ context.registerBean(SpResourceManager.class, () ->
mock(SpResourceManager.class, RETURNS_DEEP_STUBS));
context.register(StorageApiConfiguration.class,
PipelineManagementConfiguration.class, AdapterStorages.class);
context.refresh();
