openwhisk-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dube...@apache.org
Subject [incubator-openwhisk] branch master updated: Improve robustness of KafkaConsumerConnector (#3922)
Date Tue, 31 Jul 2018 20:28:07 GMT
This is an automated email from the ASF dual-hosted git repository.

dubeejw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-openwhisk.git


The following commit(s) were added to refs/heads/master by this push:
     new d847d0f  Improve robustness of KafkaConsumerConnector (#3922)
d847d0f is described below

commit d847d0f75ec2192015e3dcda19f8e5d0bc0926d5
Author: Sven Lange-Last <sven.lange-last@de.ibm.com>
AuthorDate: Tue Jul 31 22:28:03 2018 +0200

    Improve robustness of KafkaConsumerConnector (#3922)
    
    * Improve robustness of KafkaConsumerConnector
    
    * We schedule a wake-up thread once to wake up the long `poll()` call if it takes too
long. Make sure that this wake-up thread is cancelled as soon as possible once the `poll()`
has returned - no matter whether it was successful or not. Otherwise the wake-up will be remembered
by the consumer and applied on the next `poll()`. See https://kafka.apache.org/11/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#endOffsets-java.util.Collection-
    * Catch failures when closing the old consumer during re-creation. In the past, re-creation
was interrupted when closing failed and no new consumer was created.
    * Add more logging such that unexpected or error conditions no more happen unnoticed.
    
    * Address review feedback
    
    * Simplify code
---
 .../connector/kafka/KafkaConsumerConnector.scala   | 50 ++++++++++++++++++----
 1 file changed, 41 insertions(+), 9 deletions(-)

diff --git a/common/scala/src/main/scala/whisk/connector/kafka/KafkaConsumerConnector.scala
b/common/scala/src/main/scala/whisk/connector/kafka/KafkaConsumerConnector.scala
index 7111573..7e28f15 100644
--- a/common/scala/src/main/scala/whisk/connector/kafka/KafkaConsumerConnector.scala
+++ b/common/scala/src/main/scala/whisk/connector/kafka/KafkaConsumerConnector.scala
@@ -31,6 +31,7 @@ import whisk.core.connector.MessageConsumer
 import scala.collection.JavaConverters._
 import scala.concurrent.duration._
 import scala.concurrent.{blocking, ExecutionContext, Future}
+import scala.util.Failure
 
 case class KafkaConsumerConfig(sessionTimeoutMs: Long, metricFlushIntervalS: Int)
 
@@ -66,10 +67,17 @@ class KafkaConsumerConnector(
                     retry: Int = 3): Iterable[(String, Int, Long, Array[Byte])] = {
 
     // poll can be infinitely blocked in edge-cases, so we need to wakeup explicitly.
-    val wakeUpTask = actorSystem.scheduler.scheduleOnce(cfg.sessionTimeoutMs.milliseconds
+ 1.second)(consumer.wakeup())
+    val wakeUpTask = actorSystem.scheduler.scheduleOnce(cfg.sessionTimeoutMs.milliseconds
+ 1.second) {
+      consumer.wakeup()
+      logging.info(this, s"woke up consumer for topic '$topic'")
+    }
 
     try {
       val response = consumer.poll(duration.toMillis).asScala
+
+      // Cancel the scheduled wake-up task immediately.
+      wakeUpTask.cancel()
+
       val now = System.currentTimeMillis
 
       response.lastOption.foreach(record => offset = record.offset + 1)
@@ -79,21 +87,31 @@ class KafkaConsumerConnector(
         (r.topic, r.partition, r.offset, r.value)
       }
     } catch {
-      // Happens if the peek hangs.
       case _: WakeupException if retry > 0 =>
+        // Happens if the 'poll()' takes too long.
+        // This exception should occur iff 'poll()' has been woken up by the scheduled task.
+        // For this reason, it should not necessary to cancel the task. We cancel the task
+        // to be on the safe side because an ineffective `wakeup()` applies to the next
+        // consumer call that can be woken up.
+        // The scheduler is expected to safely ignore the cancellation of a task that already
+        // has been cancelled or is already complete.
+        wakeUpTask.cancel()
         logging.error(this, s"poll timeout occurred. Retrying $retry more times.")
         Thread.sleep(gracefulWaitTime.toMillis) // using Thread.sleep is okay, since `poll`
is blocking anyway
         peek(duration, retry - 1)
       case e: RetriableException if retry > 0 =>
-        logging.error(this, s"$e: Retrying $retry more times")
+        // Happens if something goes wrong with 'poll()' and 'poll()' can be retried.
         wakeUpTask.cancel()
+        logging.error(this, s"poll returned with failure. Retrying $retry more times. Exception:
$e")
         Thread.sleep(gracefulWaitTime.toMillis) // using Thread.sleep is okay, since `poll`
is blocking anyway
         peek(duration, retry - 1)
-      // Every other error results in a restart of the consumer
       case e: Throwable =>
+        // Every other error results in a restart of the consumer
+        wakeUpTask.cancel()
+        logging.error(this, s"poll returned with failure. Recreating the consumer. Exception:
$e")
         recreateConsumer()
         throw e
-    } finally wakeUpTask.cancel()
+    }
   }
 
   /**
@@ -114,8 +132,8 @@ class KafkaConsumerConnector(
     }
 
   override def close(): Unit = {
+    logging.info(this, s"closing consumer for '$topic'")
     consumer.close()
-    logging.info(this, s"closing '$topic' consumer")
   }
 
   /** Creates a new kafka consumer and subscribes to topic list if given. */
@@ -135,9 +153,19 @@ class KafkaConsumerConnector(
   }
 
   private def recreateConsumer(): Unit = {
-    val oldConsumer = consumer
-    oldConsumer.close()
-    logging.info(this, s"old consumer closed")
+    logging.info(this, s"recreating consumer for '$topic'")
+    try {
+      consumer.close()
+    } catch {
+      // According to documentation, the consumer is force closed if it cannot be closed
gracefully.
+      // See https://kafka.apache.org/11/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html
+      //
+      // For the moment, we have no special handling of 'InterruptException' - it may be
possible or even
+      // needed to re-try the 'close()' when being interrupted.
+      case t: Throwable =>
+        logging.error(this, s"failed to close old consumer while recreating: $t")
+    }
+    logging.info(this, s"old consumer closed for '$topic'")
     consumer = createConsumer(topic)
   }
 
@@ -157,6 +185,10 @@ class KafkaConsumerConnector(
           }
         }
       }
+    }.andThen {
+      case Failure(e) =>
+        // Only log level info because failed metric reporting is not critical
+        logging.info(this, s"lag metric reporting failed for topic '$topic': $e")
     }
   }
 }


Mime
View raw message