Add support for Apache Pulsar as a message transport layer

[RobertIndie]

Currently, Apache StreamPipes support 4 message protocols as the message transport layer: jms, kafka, mqtt and nats. Apache Pulsar is also a commonly used messaging protocol. I propose to add support for the Apache Pulsar.

But when I looked into the code, I found it was more complex than just adding a module. We also need to add special handling logic elsewhere. Such as

streampipes/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/DataSetGroundingSelector.java

Lines 55 to 83 in e271aa2

	if (isPrioritized(prioritizedProtocol, JmsTransportProtocol.class)) {
	outputGrounding.setTransportProtocol(makeTransportProtocol(
	BackendConfig.INSTANCE.getJmsHost(),
	BackendConfig.INSTANCE.getJmsPort(),
	topicDefinition,
	JmsTransportProtocol.class
	));
	} else if (isPrioritized(prioritizedProtocol, KafkaTransportProtocol.class)) {
	outputGrounding.setTransportProtocol(makeTransportProtocol(
	BackendConfig.INSTANCE.getKafkaHost(),
	BackendConfig.INSTANCE.getKafkaPort(),
	topicDefinition,
	KafkaTransportProtocol.class
	));
	} else if (isPrioritized(prioritizedProtocol, MqttTransportProtocol.class)) {
	outputGrounding.setTransportProtocol(makeTransportProtocol(
	BackendConfig.INSTANCE.getMqttHost(),
	BackendConfig.INSTANCE.getMqttPort(),
	topicDefinition,
	MqttTransportProtocol.class
	));
	} else if (isPrioritized(prioritizedProtocol, NatsTransportProtocol.class)) {
	outputGrounding.setTransportProtocol(makeTransportProtocol(
	BackendConfig.INSTANCE.getNatsHost(),
	BackendConfig.INSTANCE.getNatsPort(),
	topicDefinition,
	NatsTransportProtocol.class
	));
	}

I am wondering if there is a relevant guide or example PR that could help me implement this.

[dominikriemer]

Hi @RobertIndie, sounds great!
There is currently a draft (huge) PR by @tenthe which removes data sets so that it should be easier to integrate a new messaging protocol.
So it should be fine to start with a new messaging module for pulsar and then add this as supportedProtocol to the SDK.

I'll check the codebase and come back with a more detailed description!

[RobertIndie]

Update: I have created a draft PR #1576

[RobertIndie]

I have encountered an issue here.

If I enable the pulsar messaging layer, there will be a protocol match issue when editing the pipeline:

The root cause is that the expected protocol doesn't contain the PulsarTransportProtocol when checking the match of the protocols:

After I dig into the code, I find that this requirement is passed from the UI:

The supportedGrounding.transportProtocols doesn't include the Pulsar. I think I must have missed setting it somewhere.

Where can I add the pulsar to the supportedGrounding.transportProtocols?
Could someone please take a look here?

[dominikriemer]

It's a bit confusing currently as the model needs to be updated from the Install Pipeline Elements page for each installed pipeline element after registering the protocol in the Initclasss of the module. Does this work from a plain installation or do you get the same error?
I'll also have a look at the PR!

[RobertIndie]

Does this work from a plain installation or do you get the same error?

I reinstall the elements and it works. It's indeed a little confusing that the element wouldn't update automatically. Is it a point that we can improve?

[dominikriemer]

Yes, that's definitely something we can improve. But I'm not sure how a good solution could look like. Since the description of pipeline elements is stored in the core, automatically applying changes to pipeline element configurations could break existing pipelines (e.g., when removing a configuration or so).

Woudl you like to automatically update all pipeline elements upon registration or do you have any other idea how this should work?

[bossenti]

Does an instantiated pipeline still rely on the centrally managed descriptions or does it "own" a copy?

[dominikriemer]

The pipeline owns a copy: It includes a list of data processor/sink invocations which include all user-defined parameters and are created based on the description at the time of pipeline creation.
We could exclude the protocol from the model and assign the communication protocol based on the prioritized protocol at pipeline startup time. The drawback is that compatibility checking of protocols is lost (currently, protocol compatibility between pipeline elements is checked and a protocol is selected based on all matching protocols and the messaging protocol priority assigned in the UI).
On the other hand, it would certainly reduce complexity...

[bossenti]

cc #1584