Simple pulsar message listener

SimplePulsarMessageListener.java

@Service
@Slf4j
@ConditionalOnProperty(name = "consumer.enabled", havingValue = "true")
public class SimpleMessageListener implements MessageListener<byte[]> {

    private final int maxConsumeDelaySeconds;
    private static final String CONSUMER_PREFIX = "(CONSUMER) [{}]: ";

    public SimpleMessageListener(
            @Value("${consumer.max.consume.delay.seconds}") int maxConsumeDelaySeconds) {
        this.maxConsumeDelaySeconds = maxConsumeDelaySeconds;
    }

    @Override
    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
        log.info(CONSUMER_PREFIX + "Received, beginning to process ...", msg.getMessageId());
        try {
            // Sleep for a random delay to simulate delay in message consumption/processing
            TimeUnit.SECONDS.sleep((int) Math.ceil(Math.random() * maxConsumeDelaySeconds));
            consumer.acknowledge(msg);
            log.info(CONSUMER_PREFIX + "Successfully processed!", msg.getMessageId());
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.error("(CONSUMER) THREAD INTERRUPTED --- DID NOT FINISH PROCESSING!");
        } catch (PulsarClientException e) {
            log.error("(CONSUMER) Pulsar client exception", e);
            consumer.negativeAcknowledge(msg);
        }
    }
}