Merge branch 'master' into add-missing-config

Author: upgle

ansible/README.md

@@ -157,6 +157,8 @@ It will run one more component called "scheduler" and ETCD.
 You can update service providers for the scheduler as follows. 
 
 **common/scala/src/main/resources/reference.conf**
+
+If you are using ElasticSearch (recommended) then replace ```NoopDurationCheckerProvider``` with ```ElasticSearchDurationCheckerProvider``` below.
 ```
 whisk.spi {
   ArtifactStoreProvider = org.apache.openwhisk.core.database.CouchDbStoreProvider
@@ -169,27 +171,13 @@ whisk.spi {
   AuthenticationDirectiveProvider = org.apache.openwhisk.core.controller.BasicAuthenticationDirective
   InvokerProvider = org.apache.openwhisk.core.invoker.FPCInvokerReactive
   InvokerServerProvider = org.apache.openwhisk.core.invoker.FPCInvokerServer
-  DurationCheckerProvider = org.apache.openwhisk.core.scheduler.queue.ElasticSearchDurationCheckerProvider
+  DurationCheckerProvider = org.apache.openwhisk.core.scheduler.queue.NoopDurationCheckerProvider
 }
 .
 .
 .
 ```
 
-#### Configure akka dispatcher for the scheduler
-Add a new dispatcher entry as follows.
-
-**common/scala/src/main/resources/reference.conf**
-```
-  lease-service-dispatcher {
-    executor = "thread-pool-executor"
-    type = PinnedDispatcher
-  }
-.
-.
-.
-```
-
 #### Enable the scheduler
 - Make sure you enable the scheduler by configuring `scheduler_enable`.
 

ansible/group_vars/all

@@ -510,7 +510,7 @@ scheduler:
   extraEnv: "{{ scheduler_extraEnv | default({}) }}"
   dataManagementService:
     retryInterval: "{{ scheduler_dataManagementService_retryInterval | default('1 second') }}"
-  inProgressJobRetentionSecond: "{{ scheduler_inProgressJobRetentionSecond | default('20 seconds') }}"
+  inProgressJobRetention: "{{ scheduler_inProgressJobRetention | default('20 seconds') }}"
   managedFraction: "{{ scheduler_managed_fraction | default(1.0 - (scheduler_blackbox_fraction | default(__scheduler_blackbox_fraction))) }}"
   blackboxFraction: "{{ scheduler_blackbox_fraction | default(__scheduler_blackbox_fraction) }}"
   queueManager:

ansible/roles/cli/tasks/deploy.yml

@@ -35,7 +35,7 @@
   get_url:
     url: "{{ openwhisk_cli.remote.location }}/{{ openwhisk_cli.archive_name}}-{{ openwhisk_cli_tag }}-all.tgz"
     dest: "{{ nginx.confdir }}/cli_temp/{{ openwhisk_cli.archive_name }}.tgz"
-    headers: "{{ openwhisk_cli.remote.headers | default('') }}"
+    headers: "{{ openwhisk_cli.remote.headers | default('{}') }}"
   when: openwhisk_cli.installation_mode == "remote"
 
 - name: "... or Copy release archive to build directory"

ansible/roles/schedulers/tasks/deploy.yml

@@ -112,7 +112,7 @@
       "CONFIG_whisk_scheduler_protocol": "{{ scheduler.protocol }}"
       "CONFIG_whisk_scheduler_maxPeek": "{{ scheduler.maxPeek }}"
       "CONFIG_whisk_scheduler_dataManagementService_retryInterval": "{{ scheduler.dataManagementService.retryInterval }}"
-      "CONFIG_whisk_scheduler_inProgressJobRetention": "{{ scheduler.inProgressJobRetentionSecond }}"
+      "CONFIG_whisk_scheduler_inProgressJobRetention": "{{ scheduler.inProgressJobRetention }}"
       "CONFIG_whisk_scheduler_queueManager_maxSchedulingTime": "{{ scheduler.queueManager.maxSchedulingTime }}"
       "CONFIG_whisk_scheduler_queueManager_maxRetriesToGetQueue": "{{ scheduler.queueManager.maxRetriesToGetQueue }}"
       "CONFIG_whisk_scheduler_queue_idleGrace": "{{ scheduler.queue.idleGrace }}"

common/scala/src/main/scala/org/apache/openwhisk/connector/kafka/KafkaProducerConnector.scala

@@ -27,7 +27,7 @@ import pureconfig.generic.auto._
 import org.apache.openwhisk.common.{Counter, Logging, TransactionId}
 import org.apache.openwhisk.connector.kafka.KafkaConfiguration._
 import org.apache.openwhisk.core.ConfigKeys
-import org.apache.openwhisk.core.connector.{Message, MessageProducer}
+import org.apache.openwhisk.core.connector.{Message, MessageProducer, ResultMetadata}
 import org.apache.openwhisk.core.entity.{ByteSize, UUIDs}
 import org.apache.openwhisk.utils.Exceptions
 
@@ -49,17 +49,18 @@ class KafkaProducerConnector(
   override def sentCount(): Long = sentCounter.cur
 
   /** Sends msg to topic. This is an asynchronous operation. */
-  override def send(topic: String, msg: Message, retry: Int = 3): Future[RecordMetadata] = {
+  override def send(topic: String, msg: Message, retry: Int = 3): Future[ResultMetadata] = {
     implicit val transid: TransactionId = msg.transid
     val record = new ProducerRecord[String, String](topic, "messages", msg.serialize)
-    val produced = Promise[RecordMetadata]()
+    val produced = Promise[ResultMetadata]()
 
     Future {
       blocking {
         try {
           producer.send(record, new Callback {
             override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
-              if (exception == null) produced.trySuccess(metadata)
+              if (exception == null)
+                produced.trySuccess(ResultMetadata(metadata.topic(), metadata.partition(), metadata.offset()))
               else produced.tryFailure(exception)
             }
           })
@@ -72,7 +73,7 @@ class KafkaProducerConnector(
 
     produced.future.andThen {
       case Success(status) =>
-        logging.debug(this, s"sent message: ${status.topic()}[${status.partition()}][${status.offset()}]")
+        logging.debug(this, s"sent message: ${status.topic}[${status.partition}][${status.offset}]")
         sentCounter.next()
       case Failure(t) =>
         logging.error(this, s"sending message on topic '$topic' failed: ${t.getMessage}")

common/scala/src/main/scala/org/apache/openwhisk/connector/lean/LeanProducer.scala

@@ -18,13 +18,11 @@
 package org.apache.openwhisk.connector.lean
 
 import akka.actor.ActorSystem
+
 import scala.concurrent.Future
-import org.apache.kafka.clients.producer.RecordMetadata
-import org.apache.kafka.common.TopicPartition
 import org.apache.openwhisk.common.Counter
 import org.apache.openwhisk.common.Logging
-import org.apache.openwhisk.core.connector.Message
-import org.apache.openwhisk.core.connector.MessageProducer
+import org.apache.openwhisk.core.connector.{Message, MessageProducer, ResultMetadata}
 
 import java.util.concurrent.{BlockingQueue, LinkedBlockingQueue}
 import scala.collection.mutable.Map
@@ -39,15 +37,15 @@ class LeanProducer(queues: Map[String, BlockingQueue[Array[Byte]]])(implicit log
   override def sentCount(): Long = sentCounter.cur
 
   /** Sends msg to topic. This is an asynchronous operation. */
-  override def send(topic: String, msg: Message, retry: Int = 3): Future[RecordMetadata] = {
+  override def send(topic: String, msg: Message, retry: Int = 3): Future[ResultMetadata] = {
     implicit val transid = msg.transid
 
     val queue = queues.getOrElseUpdate(topic, new LinkedBlockingQueue[Array[Byte]]())
 
     Future {
       queue.put(msg.serialize.getBytes(StandardCharsets.UTF_8))
       sentCounter.next()
-      new RecordMetadata(new TopicPartition(topic, 0), -1, -1, System.currentTimeMillis(), null, -1, -1)
+      ResultMetadata(topic, 0, -1)
     }
   }
 

common/scala/src/main/scala/org/apache/openwhisk/core/WhiskConfig.scala

@@ -302,7 +302,7 @@ object ConfigKeys {
   val schedulerMaxPeek = "whisk.scheduler.max-peek"
   val schedulerQueue = "whisk.scheduler.queue"
   val schedulerQueueManager = "whisk.scheduler.queue-manager"
-  val schedulerInProgressJobRetentionSecond = "whisk.scheduler.in-progress-job-retention"
+  val schedulerInProgressJobRetention = "whisk.scheduler.in-progress-job-retention"
 
   val whiskClusterName = "whisk.cluster.name"
 

common/scala/src/main/scala/org/apache/openwhisk/core/connector/Message.scala

@@ -48,6 +48,8 @@ trait Message {
   override def toString = serialize
 }
 
+case class ResultMetadata(topic: String, partition: Int, offset: Long)
+
 case class ActivationMessage(override val transid: TransactionId,
                              action: FullyQualifiedEntityName,
                              revision: DocRevision,

common/scala/src/main/scala/org/apache/openwhisk/core/connector/MessageProducer.scala

@@ -19,15 +19,13 @@ package org.apache.openwhisk.core.connector
 
 import scala.concurrent.Future
 
-import org.apache.kafka.clients.producer.RecordMetadata
-
 trait MessageProducer {
 
   /** Count of messages sent. */
   def sentCount(): Long
 
   /** Sends msg to topic. This is an asynchronous operation. */
-  def send(topic: String, msg: Message, retry: Int = 0): Future[RecordMetadata]
+  def send(topic: String, msg: Message, retry: Int = 0): Future[ResultMetadata]
 
   /** Closes producer. */
   def close(): Unit

core/controller/src/main/scala/org/apache/openwhisk/core/loadBalancer/CommonLoadBalancer.scala

@@ -23,7 +23,6 @@ import java.util.concurrent.atomic.LongAdder
 
 import akka.actor.ActorSystem
 import akka.event.Logging.InfoLevel
-import org.apache.kafka.clients.producer.RecordMetadata
 import pureconfig._
 import pureconfig.generic.auto._
 import org.apache.openwhisk.common.LoggingMarkers._
@@ -189,7 +188,7 @@ abstract class CommonLoadBalancer(config: WhiskConfig,
   /** 3. Send the activation to the invoker */
   protected def sendActivationToInvoker(producer: MessageProducer,
                                         msg: ActivationMessage,
-                                        invoker: InvokerInstanceId): Future[RecordMetadata] = {
+                                        invoker: InvokerInstanceId): Future[ResultMetadata] = {
     implicit val transid: TransactionId = msg.transid
 
     val topic = s"${Controller.topicPrefix}invoker${invoker.toInt}"
@@ -206,7 +205,7 @@ abstract class CommonLoadBalancer(config: WhiskConfig,
         transid.finished(
           this,
           start,
-          s"posted to ${status.topic()}[${status.partition()}][${status.offset()}]",
+          s"posted to ${status.topic}[${status.partition}][${status.offset}]",
           logLevel = InfoLevel)
       case Failure(_) => transid.failed(this, start, s"error on posting to topic $topic")
     }