Simple Pulsar queue consumer

SimplePulsarQueueConsumer.java

@Configuration
@ConditionalOnProperty(name = "consumer.enabled", havingValue = "true")
@Slf4j
public class PulsarConsumerConfiguration {

    private final PulsarClient client;
    private final List<Consumer> consumers;
    private final SimpleMessageListener msgListener;

    public PulsarConsumerConfiguration(
            @Value("${pulsar.url}") String pulsarUrl,
            @Value("${pulsar.topic.name}") String topicName,
            @Value("${pulsar.subscription.name}") String subscriptionName,
            SimpleMessageListener messageListener
    ) throws PulsarClientException {
        this.msgListener = messageListener;
        client = PulsarClient.builder()
                .serviceUrl(pulsarUrl)
                .build();
        log.info("(CONSUMER) Connected to Pulsar client");
        consumers = new ArrayList<>();
        Consumer<byte[]> consumer = client.newConsumer()
                .subscriptionName(subscriptionName)
                .subscriptionType(SubscriptionType.Key_Shared)
                .topic(topicName)
                .receiverQueueSize(5)
                .messageListener(messageListener)
                .ackTimeout(10, TimeUnit.SECONDS)
                .subscribe();

        consumers.add(consumer);

        log.info("(CONSUMER) Subscribed to topic {}", topicName);
    }

}