Merge pull request #26 from majusko/feature/issue-25-message-error-handling

Feature/issue 25 message error handling

Author: majusko

pom.xml

@@ -23,6 +23,7 @@
 		<testcontainers.version>1.14.3</testcontainers.version>
 		<junit-jupiter.version>5.6.2</junit-jupiter.version>
 		<awaitility.version>4.0.3</awaitility.version>
+		<reactor.core.version>3.3.6.RELEASE</reactor.core.version>
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 	</properties>
 
@@ -38,6 +39,13 @@
 			<version>${apache.pulsar.client.version}</version>
 		</dependency>
 
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<version>${reactor.core.version}</version>
+		</dependency>
+
+		<!--	TESTS	-->
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
@@ -63,7 +71,6 @@
 			<scope>test</scope>
 		</dependency>
 
-        <!--	Pulsar integration tests	-->
 		<dependency>
 			<groupId>org.testcontainers</groupId>
 			<artifactId>testcontainers</artifactId>

src/main/java/io/github/majusko/pulsar/consumer/ConsumerBuilder.java

@@ -2,12 +2,16 @@
 
 import io.github.majusko.pulsar.collector.ConsumerCollector;
 import io.github.majusko.pulsar.collector.ConsumerHolder;
+import io.github.majusko.pulsar.error.FailedMessage;
+import io.github.majusko.pulsar.error.exception.ConsumerInitException;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.stereotype.Component;
+import reactor.core.Disposable;
+import reactor.core.publisher.EmitterProcessor;
 
 import javax.annotation.PostConstruct;
 import java.lang.reflect.Method;
@@ -18,6 +22,7 @@
 @DependsOn({"pulsarClient", "consumerCollector"})
 public class ConsumerBuilder {
 
+    private final EmitterProcessor<FailedMessage> exceptionEmitter = EmitterProcessor.create();
     private final ConsumerCollector consumerCollector;
     private final PulsarClient pulsarClient;
 
@@ -53,15 +58,19 @@ private Consumer<?> subscribe(String name, ConsumerHolder holder) {
                         consumer.acknowledge(msg);
                     } catch (Exception e) {
                         consumer.negativeAcknowledge(msg);
-                        throw new RuntimeException("TODO Custom Exception!", e);
+                        exceptionEmitter.onNext(new FailedMessage(e, consumer, msg));
                     }
                 }).subscribe();
         } catch (PulsarClientException e) {
-            throw new RuntimeException("TODO Custom Exception!", e);
+            throw new ConsumerInitException("Failed to init consumer.", e);
         }
     }
 
     public List<Consumer> getConsumers() {
         return consumers;
     }
+
+    public Disposable onError(java.util.function.Consumer<? super FailedMessage> consumer) {
+        return exceptionEmitter.subscribe(consumer);
+    }
 }

src/main/java/io/github/majusko/pulsar/error/FailedMessage.java

@@ -0,0 +1,28 @@
+package io.github.majusko.pulsar.error;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+
+public class FailedMessage {
+    private final Exception exception;
+    private final Consumer<?> consumer;
+    private final Message<?> message;
+
+    public FailedMessage(Exception exception, Consumer<?> consumer, Message<?> message) {
+        this.exception = exception;
+        this.consumer = consumer;
+        this.message = message;
+    }
+
+    public Exception getException() {
+        return exception;
+    }
+
+    public Consumer<?> getConsumer() {
+        return consumer;
+    }
+
+    public Message<?> getMessage() {
+        return message;
+    }
+}

src/main/java/io/github/majusko/pulsar/error/exception/ConsumerInitException.java

@@ -0,0 +1,7 @@
+package io.github.majusko.pulsar.error.exception;
+
+public class ConsumerInitException extends RuntimeException {
+    public ConsumerInitException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

src/main/java/io/github/majusko/pulsar/error/exception/ProducerInitException.java

@@ -0,0 +1,11 @@
+package io.github.majusko.pulsar.error.exception;
+
+public class ProducerInitException extends RuntimeException {
+    public ProducerInitException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public ProducerInitException(String message) {
+        super(message);
+    }
+}

src/main/java/io/github/majusko/pulsar/producer/ProducerCollector.java

@@ -3,6 +3,7 @@
 import io.github.majusko.pulsar.annotation.PulsarProducer;
 import io.github.majusko.pulsar.collector.ProducerHolder;
 import io.github.majusko.pulsar.constant.Serialization;
+import io.github.majusko.pulsar.error.exception.ProducerInitException;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
@@ -19,7 +20,7 @@ public class ProducerCollector implements BeanPostProcessor {
 
     private final PulsarClient pulsarClient;
 
-    private Map<String, Producer> producers = new ConcurrentHashMap<>();
+    private final Map<String, Producer> producers = new ConcurrentHashMap<>();
 
     public ProducerCollector(PulsarClient pulsarClient) {
         this.pulsarClient = pulsarClient;
@@ -29,7 +30,7 @@ public ProducerCollector(PulsarClient pulsarClient) {
     public Object postProcessBeforeInitialization(Object bean, String beanName) {
         final Class<?> beanClass = bean.getClass();
 
-        if(beanClass.isAnnotationPresent(PulsarProducer.class) && bean instanceof PulsarProducerFactory) {
+        if (beanClass.isAnnotationPresent(PulsarProducer.class) && bean instanceof PulsarProducerFactory) {
             producers.putAll(((PulsarProducerFactory) bean).getTopics().entrySet().stream()
                 .map($ -> new ProducerHolder($.getKey(), $.getValue().left, $.getValue().right))
                 .collect(Collectors.toMap(ProducerHolder::getTopic, this::buildProducer)));
@@ -48,16 +49,16 @@ private Producer<?> buildProducer(ProducerHolder holder) {
             return pulsarClient.newProducer(getSchema(holder))
                 .topic(holder.getTopic())
                 .create();
-        } catch(PulsarClientException e) {
-            throw new RuntimeException("TODO Custom Exception!", e);
+        } catch (PulsarClientException e) {
+            throw new ProducerInitException("Failed to init producer.", e);
         }
     }
 
     private <T> Schema<?> getSchema(ProducerHolder holder) throws RuntimeException {
         if (holder.getSerialization().equals(Serialization.JSON)) {
             return Schema.JSON(holder.getClazz());
         }
-        throw new RuntimeException("TODO custom runtime exception");
+        throw new ProducerInitException("Unknown producer schema.");
     }
 
     Map<String, Producer> getProducers() {

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,12 +70,11 @@ 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);
+        final Consumer<?> consumer = consumers.stream().filter( $-> $.getTopic().equals("topic-one")).findFirst().orElseThrow(Exception::new);
 
         Assertions.assertNotNull(consumer);
-        Assertions.assertEquals("topic-one", consumer.getTopic());
     }
 
     @Test
@@ -91,11 +94,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);
     }