Fixing Snapshots for Sink and Source
@@ -344,7 +344,6 @@ public class DefinitionParserHelper {
sourceType, sourceAnnotation, SiddhiConstants.NAMESPACE_SOURCE);
Source source = (Source) SiddhiClassLoader.loadExtensionImplementation(sourceExtension,
SourceExecutorExtensionHolder.getInstance(siddhiAppContext));
- siddhiAppContext.getSnapshotService().addSnapshotable(source.getStreamDefinition().getId(), source);
// load input mapper extension
Extension mapperExtension = constructExtension(streamDefinition, SiddhiConstants.ANNOTATION_MAP,
@@ -373,6 +372,8 @@ public class DefinitionParserHelper {
.generateConfigReader(sourceExtension.getNamespace(), sourceExtension.getName()),
streamDefinition, siddhiAppContext);
+ siddhiAppContext.getSnapshotService().addSnapshotable(source.getStreamDefinition().getId(), source);
+
List<Source> eventSources = eventSourceMap.get(streamDefinition.getId());
if (eventSources == null) {
eventSources = new ArrayList<>();
@@ -474,8 +475,6 @@ public class DefinitionParserHelper {
} else {
sink = createSink(sinkExtension, siddhiAppContext);
}
- siddhiAppContext.getSnapshotService().addSnapshotable(sink.getStreamDefinition().getId(), sink);
-
if (supportedDynamicOptions == null) {
supportedDynamicOptions = sink.getSupportedDynamicOptions();
}
@@ -540,6 +539,8 @@ public class DefinitionParserHelper {
sink.getMapper().setGroupDeterminer(groupDeterminer);
}
+ siddhiAppContext.getSnapshotService().addSnapshotable(sink.getStreamDefinition().getId(), sink);
+
List<Sink> eventSinks = eventSinkMap.get(streamDefinition.getId());
if (eventSinks == null) {
eventSinks = new ArrayList<>();
