Apache Pulsar Adapter

The Apache Pulsar connectivity solution is implemented as a pair of global Java operators that allow a Spotfire Streaming application to connect to an Apache Pulsar broker and exchange messages with it.

Separate instances of the Pulsar operators can be made to share a connection to the broker. Each operator lists the connection configurations available in its Connection Definition adapter property (see Properties: Operator Properties Tab). The list is generated from the entries specified in a dedicated section of the application's adapter configuration file. The following example is very similar to the Pulsar.conf file used in the Pulsar sample:

name = "Pulsar.conf"
type = "com.tibco.ep.streambase.configuration.adapter"
version = "1.0.0"
configuration = {

    // An adapter group type defines a collection of EventFlow adapter configurations, 
    // indexed by adapter type.
    AdapterGroup = {

        // A collection of EventFlow adapter configurations, indexed by adapter type. 
        // This key is required and must contain at least one configuration.
        adapters = {

            // The root section for an adapter configuration.
            pulsar = {

                // Section list. This key is optional and has no default value.
                sections = [ 

                    // A configuration for an adapter named section.
                    {

                        // Section name. The value does not have to be unique; 
                        // that is, you can have multiple sections with the same name
                        // in the same array of sections. This key is required.
                        name = "broker-connection"

                        // Section property bag. All values must be strings. 
                        // This key is optional and has no default value.
                        settings = {

                            broker-uri = "pulsar://localhost:6650"

                            // This value may be anything you like -- 
                            // it's the value that will appear in each operator's
                            // Connection Definition drop-down to designate this entry.
                            id = "Test Broker 1"
                            
                            // One of: None, HTTPBasic, TLSAuth, TLSEncryption, TLSBoth 
                            // (i.e. TLSAuth+TLSEncryption)
                            authentication-type = "None"

                            // HTTPBasic auth credentials. Passwords may be enciphered 
                            // using the "epadmin encrypt" family of commands.

                            username = "myusername"
                            password = "mypassword"

                            // TLS auth and encryption settings. Passwords may be enciphered.

                            //use-keystore-tls = "false"
                            //tls-keystore-path = ""
                            //tls-keystore-password = ""
                            //tls-truststore-path = ""
                            //tls-truststore-password = ""
                            //enable-hostname-verification = "false"
                            //allow-tls-insecure-connection = "false"
                        }
                    }
                    // A configuration for another adapter named section.
                    {

                        // Section name. The value does not have to be unique; 
                        // that is, you can have multiple sections with the same name
                        // in the same array of sections. This key is required.
                        name = "broker-connection"

                        // Section property bag. All values must be strings. 
                        // This key is optional and has no default value.
                        settings = {

                            broker-uri = "pulsar://example.com:6650"

                            // This value may be anything you like -- 
                            // it's the value that will appear in each operator's
                            // Connection Definition drop-down to designate this entry.
                            id = "Test Broker 2"
                            
                            // One of: None, HTTPBasic, TLSAuth, TLSEncryption, TLSBoth
                            authentication-type = "None"

                            // HTTPBasic auth credentials. 
                            // Passwords may be enciphered using the "epadmin encrypt" family of commands.

                            username = "myusername"
                            password = "mypassword"

                            // TLS auth and encryption settings. Passwords may be enciphered.

                            //use-keystore-tls = "false"
                            //tls-keystore-path = ""
                            //tls-keystore-password = ""
                            //tls-truststore-path = ""
                            //tls-truststore-password = ""
                            //enable-hostname-verification = "false"
                            //allow-tls-insecure-connection = "false"
                        }
                    }
                ]
            }
        }
    }
}

In each section named broker-connection, the value of the id setting identifies this particular section (in other words, this particular connection definition) and will be listed in the Operator Properties tab's Adapter Configuration drop-down setting in each operator's Properties view. Multiple operators can be made to share a single connection by using the same value for this Adapter Configuration setting.

A best practice is to define your connections before placing operator instances on the canvas, so that the list is already available in the Properties view and the operators can be configured right away.

name = "pulsar-adapter"
version = 1.0.0
type = "com.tibco.ep.streambase.configuration.sbengine"

//
// Streaming engine configuration for Pulsar adapter.
//
configuration =
{
    StreamBaseEngine =
    {
        // This --add-opens directive is needed when running under Java 17 or later 
        // due to Apache Pulsar's use of reflection.
        jvmArgs =
        [
            "--add-opens=java.base/sun.net=ALL-UNNAMED"                            
        ]
    }
}

This section describes the properties you can set for a Pulsar operator, using the various tabs of the Properties view in StreamBase Studio.

Each operator has a Status output port used to issue various informational and error messages:

The Spotfire Streaming installation comes with a sample demonstrating the use of this adapter. To load the sample in Streaming Studio, select File>Import Samples and Community Content and look under the Messaging Adapters category for an entry called Pulsar Adapter.

Back to Top ^