Added test for error handler feature.

Author: majusko

src/test/java/io/github/majusko/pulsar/PulsarJavaSpringBootStarterApplicationTests.java

@@ -24,6 +24,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.awaitility.Awaitility.await;
 
@@ -44,6 +45,9 @@ class PulsarJavaSpringBootStarterApplicationTests {
     @Autowired
     private PulsarTemplate<MyMsg> producer;
 
+    @Autowired
+    private PulsarTemplate<String> producerForError;
+
     @Container
     static PulsarContainer pulsarContainer = new PulsarContainer();
 
@@ -66,7 +70,7 @@ void testProducerSendMethod() throws PulsarClientException {
     void testConsumerRegistration1() throws Exception {
         final List<Consumer<?>> consumers = consumerBuilder.getConsumers();
 
-        Assertions.assertEquals(1, consumers.size());
+        Assertions.assertEquals(2, consumers.size());
 
         final Consumer<?> consumer = consumers.stream().findFirst().orElseThrow(Exception::new);
 
@@ -91,11 +95,29 @@ void testProducerRegistration() {
 
         final Map<String, ImmutablePair<Class<?>, Serialization>> topics = producerFactory.getTopics();
 
-        Assertions.assertEquals(2, topics.size());
+        Assertions.assertEquals(3, topics.size());
 
         final Set<String> topicNames = new HashSet<>(topics.keySet());
 
         Assertions.assertTrue(topicNames.contains("topic-one"));
         Assertions.assertTrue(topicNames.contains("topic-two"));
     }
+
+    @Test
+    void testMessageErrorHandling() throws PulsarClientException {
+        final AtomicBoolean receivedError = new AtomicBoolean(false);
+        final String messageToSend = "This message will never arrive.";
+
+        producerForError.send("topic-for-error", messageToSend);
+
+        consumerBuilder.onError(($) -> {
+            Assertions.assertEquals($.getConsumer().getTopic(), "topic-for-error");
+            Assertions.assertEquals($.getMessage().getValue(), messageToSend);
+            Assertions.assertNotNull($.getException());
+
+            receivedError.set(true);
+        });
+
+        await().untilTrue(receivedError);
+    }
 }

src/test/java/io/github/majusko/pulsar/TestConsumers.java

@@ -17,4 +17,8 @@ public void topicOneListener(MyMsg myMsg) {
         Assertions.assertNotNull(myMsg);
         mockTopicListenerReceived.set(true);
     }
+
+    @PulsarConsumer(topic = "topic-for-error", clazz = String.class, serialization = Serialization.JSON)
+    public void topicForErrorListener(Integer myMsg) {
+    }
 }

src/test/java/io/github/majusko/pulsar/TestProducerConfiguration.java

@@ -11,6 +11,7 @@ public class TestProducerConfiguration {
     @Bean
     public ProducerFactory producerFactory() {
         return new ProducerFactory()
+            .addProducer("topic-for-error", String.class)
             .addProducer("topic-one", MyMsg.class)
             .addProducer("topic-two", MyMsg2.class, Serialization.JSON);
     }