Spring boot starter for Apache Pulsar

Quick Start

Simple start consist only from 3 simple steps.

1. Add Maven dependency

<dependency>
  <groupId>io.github.majusko</groupId>
  <artifactId>pulsar-java-spring-boot-starter</artifactId>
  <version>${version}</version>
</dependency>

2. Configure Producer

Create your configuration class with all producers you would like to register.

@Configuration
public class TestProducerConfiguration {

    @Bean
    public ProducerFactory producerFactory() {
        return new ProducerFactory()
            .addProducer("my-topic", MyMsg.class)
            .addProducer("other-topic", String.class);
    }
}

Use registered producers by simply injecting the PulsarTemplate into your service.

@Service
class MyProducer {

	@Autowired
	private PulsarTemplate<MyMsg> producer;

	void send(MyMsg msg) {
		producer.send("my-topic", msg);
	}
}

3. Configure Consumer

Annotate your service method with @PulsarConsumer annotation.

@Service
class MyConsumer {
    
    @PulsarConsumer(topic="my-topic", clazz=MyMsg.class)
    void consume(MyMsg msg) { 
        producer.send(TOPIC, msg); 
    }
}

Example project

Java Pulsar Example Project

Documentation

Configuration

Default configuration:

pulsar.serviceUrl=pulsar://localhost:6650
pulsar.ioThreads=10
pulsar.listenerThreads=10
pulsar.isEnableTcpNoDelay=false
pulsar.keepAliveIntervalSec=20
pulsar.connectionTimeoutSec=10
pulsar.operationTimeoutSec=15
pulsar.startingBackoffIntervalMs=100
pulsar.maxBackoffIntervalSec=10
pulsar.consumerNameDelimiter=

Properties explained:

• pulsar.serviceUrl - URL used to connect to pulsar cluster.
• pulsar.ioThreads - Number of threads to be used for handling connections to brokers.
• pulsar.listenerThreads - Set the number of threads to be used for message listeners/subscribers.
• pulsar.isEnableTcpNoDelay - Whether to use TCP no-delay flag on the connection, to disable Nagle algorithm.
• pulsar.keepAliveInterval - Keep alive interval for each client-broker-connection.
• pulsar.connectionTimeoutSec - duration of time to wait for a connection to a broker to be established. If the duration passes without a response from the broker, the connection attempt is dropped.
• pulsar.operationTimeoutSec - Operation timeout.
• pulsar.startingBackoffIntervalMs - Duration of time for a backoff interval (Retry algorithm).
• pulsar.maxBackoffIntervalSec - The maximum duration of time for a backoff interval (Retry algorithm).
• pulsar.consumerNameDelimiter - Consumer names are connection of bean name and method with a delimiter. By default there is no delimiter and words are connected together.

Contributing

All contributors are welcome. If you never contributed to the open-source, start with reading the Github Flow.

Roadmap task

1. Pick a task from simple roadmap in Projects section.
2. Create a pull request with reference (url) to the task inside the Projects section.
3. Rest and enjoy the great feeling of being a contributor.

Hotfix

1. Create an issue
2. Create a pull request with reference to the issue
3. Rest and enjoy the great feeling of being a contributor.