Aligning pipeline element & adapter API

[dominikriemer]

Hi,

now that we have a clean API to create custom adapters and have switched to a one-class approach for developing pipeline elements, I'd like to discuss further alignment of both APIs.

In PR #1633, I added several improvements to the pipeline element API (removed obsolete classes, replaced all classes in the SDK with interfaces and did some code reorganization). One thing I always wanted to do but which wasn't possible so far is to decleare pipeline elements using an interface instead of extending the StreamPipesDataProcessor class.

So the new approach looks like this:

New Interface to create data processors and sinks

public interface IStreamPipesDataProcessor {
  IDataProcessorConfiguration declareConfig();

  void onPipelineStarted(IDataProcessorParameters params,
                         SpOutputCollector collector,
                         EventProcessorRuntimeContext runtimeContext);

  void onEvent(Event event,
               SpOutputCollector collector);

  void onPipelineStopped();
}

This new API is closer to the adapter interface (with methods onAdapterStarted and onAdapterStopped) and hopefully eases the development of pipeline elements for new users.

The new design should also ease testability since we can more easily create mocks of the required interfaces.

Runtime layer

I also refactored the runtime layer. A drawback of the existing design was that a pipeline element implementation was bound to a specific runtime (e.g., standalone, kafka streams). With the new design, we should be able to support multiple runtimes with the same pipeline element implementation. To register a runtime, the following snippet needs to be added to the ServiceDefinition:

 @Override
  public SpServiceDefinition provideServiceDefinition() {
     return SpServiceDefinitionBuilder
        .create("ID", "title", "description", 8090)
        .registerRuntimeProvider(new StandaloneStreamPipesRuntimeProvider())
        .registerPipelineElements()
         ...
  }

So the runtimes can be directly registered. I did a quick test with the kafka streams wrapper by replacing the StandaloneStreamPipesRuntimeProvider with the ``KafkaStreamsRuntimeProvider``` which then runs all pipeline elements with Kafka Streams :-), although not yet perfectly. But this could be also interesting to e.g. run pipeline elements as Pulsar functions as well.

Compatibility

To ensure compatibility with existing classes, I refactored the StreamPipesDataProcessor class which we currently use so that implementations still work. Some imports would need to be updated for extensions that are not part of StreamPipes since I moved several interfaces to the streampipes-extensions-api module for consistency reasons.

Problems

Although I refactored the Flink wrapper as well so that the project compiles, I wasn't able to test it since our current Java version is 17 and Flink only supports 11. I'll create a separate discussion for this.

What do you think of this approach? Do you have any feedback or suggestions what we can further improve here?

[bossenti]

So just to get you right, the proposed changes are already implemented in #1633, correct? 😉

I really like the new interface for pipeline elements. It looks pretty clean and simple. In addition, I like to have one "style" of implementing both adapters & pipeline elements (now via interfaces).

The proposed improvement for the runtime layer sounds really promising 🎉

In terms of compatibility, we will have some breaking changes with the refactoring of the adapters in the next release.
So I wouldn't mind if we take this chance and do this refactoring here as well.
Anyhow, we should document the changes well and ensure that we communicate them accordingly.
Nevertheless, I appreciate that you made the effort to maintain the StreamPipesDataProcessor backwards compatible.
So we could then given another 2-3 releases before removing it entirely.

[dominikriemer]

Yes, the changes are in #1633. If someone has any idea how to further improve this, we can still merge it in the adapter branch.

[tenthe]

Hello @dominikriemer,
I really like the new design of the interface.
Maybe we can unify the names of the parameters used in the methods as well, what do you think?
For the StreamPipesAdapter we have all prefixed with an I.