diff --git a/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaJsonMessagePoller.java b/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaJsonMessagePoller.java
index 3597d7a..e437ceb 100644
--- a/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaJsonMessagePoller.java
+++ b/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaJsonMessagePoller.java
@@ -69,29 +69,30 @@
 
     public void run() {
         LOG.info("Start JSON poller for handler {}", handler);
-        try {
-            while(running) {
-                consume();
-            }
-        } catch (WakeupException e) {
-            if (running) {
-                LOG.error("Waked up while running {}", e.getMessage(), e);
-                throw e;
-            } else {
+        while(running) {
+            try {
+                consumer.poll(ofHours(1))
+                    .forEach(this::handleRecord);
+            } catch (WakeupException e) {
                 LOG.debug("Waked up while stopping {}", e.getMessage(), e);
+                running = false;
+            } catch(Exception e) {
+                eventSender.send(e);
+                LOG.error("Exception during recieve: {}", e.getMessage(), e);
+                sleepAfterError();
+                // Continue as KafkaConsumer should handle the error transparently
             }
-        } catch(Throwable t) {
-            LOG.error(format("Catch Throwable %s closing consumer", t.getMessage()), t);
-            throw t;
-        } finally {
-            consumer.close();
         }
+        consumer.close();
         LOG.info("Stop JSON poller for handler {}", handler);
     }
-
-    private void consume() {
-        consumer.poll(ofHours(1))
-                .forEach(this::handleRecord);
+    
+    private void sleepAfterError() {
+        try {
+            Thread.sleep(10000);
+        } catch (InterruptedException e1) {
+            Thread.currentThread().interrupt();
+        }
     }
 
     private void handleRecord(ConsumerRecord<String, String> record) {
diff --git a/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaMessagePoller.java b/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaMessagePoller.java
index 3e20d75..52ae1a6 100644
--- a/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaMessagePoller.java
+++ b/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaMessagePoller.java
@@ -77,27 +77,32 @@
 
     public void run() {
         LOG.info("Start poller for types {}", types);
-        try {
-            while(running) {
+        while(running) {
+            try {
                 consumer.poll(ofHours(1))
-                    .forEach(this::handleRecord);
+                .forEach(this::handleRecord);
+            } catch (WakeupException e) {
+                LOG.debug("Waked up {}", e.getMessage(), e);
+                this.running = false;
+            } catch(Exception e) {
+                eventSender.send(e);
+                LOG.error("Exception while receiving from kafka: {}", e.getMessage(), e);
+                sleepAfterError();
+                // Continue as KafkaConsumer should handle the error transparently
             }
-        } catch (WakeupException e) {
-            if (running) {
-                LOG.error("Waked up while running {}", e.getMessage(), e);
-                throw e;
-            } else {
-                LOG.debug("Waked up while stopping {}", e.getMessage(), e);
-            }
-        } catch(Throwable t) {
-            LOG.error(format("Catch Throwable %s closing consumer", t.getMessage()), t);
-            throw t;
-        } finally {
-            consumer.close();
         }
+        consumer.close();
         LOG.info("Stop poller for types {}", types);
     }
 
+    private void sleepAfterError() {
+        try {
+            Thread.sleep(10000);
+        } catch (InterruptedException e1) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
     private void handleRecord(ConsumerRecord<String, byte[]> record) {
         getHandler(record)
             .ifPresent(handler->handleRecord(handler, record));
@@ -125,7 +130,6 @@
             ByteString payload = ByteString.copyFrom(record.value());
             handler.handle(info, payload);
         } catch (Exception e) {
-            eventSender.send(e);
             String msg = format("Error consuming message for types %s", types);
             LOG.warn(msg);
         }