You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@openwhisk.apache.org by GitBox <gi...@apache.org> on 2018/04/03 06:22:39 UTC
[GitHub] markusthoemmes closed pull request #3111: Replace our own clients with akka-streams-kafka.
markusthoemmes closed pull request #3111: Replace our own clients with akka-streams-kafka.
URL: https://github.com/apache/incubator-openwhisk/pull/3111
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/.gitignore b/.gitignore
index bfb7f48f58..dc9e7a3b03 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,6 +1,7 @@
# Whisk
nginx.conf
whisk.properties
+/tests/src/test/resources/application.conf
default.props
.ant-targets-build.xml
diff --git a/ansible/roles/controller/tasks/deploy.yml b/ansible/roles/controller/tasks/deploy.yml
index ca1aa2a41d..5156cf16c6 100644
--- a/ansible/roles/controller/tasks/deploy.yml
+++ b/ansible/roles/controller/tasks/deploy.yml
@@ -51,7 +51,7 @@
"WHISK_VERSION_DATE": "{{ whisk.version.date }}"
"WHISK_VERSION_BUILDNO": "{{ docker.image.tag }}"
- "KAFKA_HOSTS": "{{ kafka_connect_string }}"
+ "CONFIG_whisk_kafka_hosts": "{{ kafka_connect_string }}"
"CONFIG_whisk_kafka_replicationFactor": "{{ kafka.replicationFactor | default() }}"
"CONFIG_whisk_kafka_topics_cacheInvalidation_retentionBytes": "{{ kafka_topics_cacheInvalidation_retentionBytes | default() }}"
"CONFIG_whisk_kafka_topics_cacheInvalidation_retentionMs": "{{ kafka_topics_cacheInvalidation_retentionMS | default() }}"
@@ -102,7 +102,7 @@
"CONFIG_kamon_statsd_port": "{{ metrics.kamon.port }}"
"CONFIG_whisk_spi_LogStoreProvider": "{{ userLogs.spi }}"
-
+
"CONFIG_logback_log_level": "{{ controller.loglevel }}"
volumes:
- "{{ whisk_logs_dir }}/controller{{ groups['controllers'].index(inventory_hostname) }}:/logs"
diff --git a/ansible/roles/invoker/tasks/deploy.yml b/ansible/roles/invoker/tasks/deploy.yml
index 3da0b30ef2..ca27ba3f76 100644
--- a/ansible/roles/invoker/tasks/deploy.yml
+++ b/ansible/roles/invoker/tasks/deploy.yml
@@ -117,7 +117,7 @@
-e INVOKER_OPTS='{{ invoker.arguments }}'
-e COMPONENT_NAME='invoker{{ groups['invokers'].index(inventory_hostname) }}'
-e PORT='8080'
- -e KAFKA_HOSTS='{{ kafka_connect_string }}'
+ -e CONFIG_whisk_kafka_hosts='{{ kafka_connect_string }}'
-e CONFIG_whisk_kafka_replicationFactor='{{ kafka.replicationFactor | default() }}'
-e CONFIG_whisk_kafka_topics_invoker_retentionBytes='{{ kafka_topics_invoker_retentionBytes | default() }}'
-e CONFIG_whisk_kafka_topics_invoker_retentionMs='{{ kafka_topics_invoker_retentionMS | default() }}'
diff --git a/ansible/tasks/writeWhiskProperties.yml b/ansible/tasks/writeWhiskProperties.yml
index 4f5fc14e20..be3cdf3eca 100644
--- a/ansible/tasks/writeWhiskProperties.yml
+++ b/ansible/tasks/writeWhiskProperties.yml
@@ -6,3 +6,8 @@
template:
src: whisk.properties.j2
dest: "{{ openwhisk_home }}/whisk.properties"
+
+- name: write test's application conf overrides
+ template:
+ src: "{{ openwhisk_home }}/tests/src/test/resources/application.conf.j2"
+ dest: "{{ openwhisk_home }}/tests/src/test/resources/application.conf"
\ No newline at end of file
diff --git a/common/scala/build.gradle b/common/scala/build.gradle
index 2c04731bbf..cf3a38568f 100644
--- a/common/scala/build.gradle
+++ b/common/scala/build.gradle
@@ -22,6 +22,7 @@ dependencies {
compile 'com.typesafe.akka:akka-http-spray-json_2.11:10.0.10'
compile 'com.lightbend.akka:akka-stream-alpakka-file_2.11:0.15'
+ compile 'com.typesafe.akka:akka-stream-kafka_2.11:0.18'
compile 'ch.qos.logback:logback-classic:1.2.3'
compile 'org.slf4j:jcl-over-slf4j:1.7.25'
diff --git a/common/scala/src/main/resources/application.conf b/common/scala/src/main/resources/application.conf
index 8af1d3d3d2..6c08cd2336 100644
--- a/common/scala/src/main/resources/application.conf
+++ b/common/scala/src/main/resources/application.conf
@@ -13,6 +13,30 @@ akka.http {
}
}
+akka.kafka {
+ consumer {
+ wakeup-timeout = 10 seconds
+
+ kafka-clients {
+ bootstrap.servers = ${whisk.kafka.hosts}
+ auto.offset.reset = "earliest"
+ session.timeout.ms = 30000
+ heartbeat.interval.ms = 10000
+ enable.auto.commit = true
+ auto.commit.interval.ms = 10000
+ max.poll.interval.ms = 300000
+ fetch.max.wait.ms = 20
+ }
+ }
+ producer {
+ kafka-clients {
+ bootstrap.servers = ${whisk.kafka.hosts}
+ acks = 1
+ }
+ }
+
+}
+
#kamon related configuration
kamon {
diff --git a/common/scala/src/main/scala/whisk/connector/kafka/KafkaConsumerConnector.scala b/common/scala/src/main/scala/whisk/connector/kafka/KafkaConsumerConnector.scala
deleted file mode 100644
index a1116d8e69..0000000000
--- a/common/scala/src/main/scala/whisk/connector/kafka/KafkaConsumerConnector.scala
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package whisk.connector.kafka
-
-import java.util.Properties
-
-import scala.collection.JavaConversions.iterableAsScalaIterable
-import scala.collection.JavaConversions.seqAsJavaList
-import scala.concurrent.duration.Duration
-import scala.concurrent.duration.DurationInt
-import scala.concurrent.duration.FiniteDuration
-
-import org.apache.kafka.clients.consumer.ConsumerConfig
-import org.apache.kafka.clients.consumer.KafkaConsumer
-import org.apache.kafka.common.serialization.ByteArrayDeserializer
-
-import whisk.common.Logging
-import whisk.core.connector.MessageConsumer
-
-class KafkaConsumerConnector(kafkahost: String,
- groupid: String,
- topic: String,
- override val maxPeek: Int = Int.MaxValue,
- readeos: Boolean = true,
- sessionTimeout: FiniteDuration = 30.seconds,
- autoCommitInterval: FiniteDuration = 10.seconds,
- maxPollInterval: FiniteDuration = 5.minutes)(implicit logging: Logging)
- extends MessageConsumer {
-
- /**
- * Long poll for messages. Method returns once message are available but no later than given
- * duration.
- *
- * @param duration the maximum duration for the long poll
- */
- override def peek(duration: Duration = 500.milliseconds) = {
- val records = consumer.poll(duration.toMillis)
- records map { r =>
- (r.topic, r.partition, r.offset, r.value)
- }
- }
-
- /**
- * Commits offsets from last poll.
- */
- def commit() = consumer.commitSync()
-
- override def close() = {
- logging.info(this, s"closing '$topic' consumer")
- }
-
- private def getProps: Properties = {
- val props = new Properties
- props.put(ConsumerConfig.GROUP_ID_CONFIG, groupid)
- props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkahost)
- props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout.toMillis.toString)
- props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, (sessionTimeout.toMillis / 3).toString)
- props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true.toString)
- props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval.toMillis.toString)
- props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPeek.toString)
- props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, if (!readeos) "latest" else "earliest")
- props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollInterval.toMillis.toString)
-
- // This value controls the server-side wait time which affects polling latency.
- // A low value improves latency performance but it is important to not set it too low
- // as that will cause excessive busy-waiting.
- props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "20")
-
- props
- }
-
- /** Creates a new kafka consumer and subscribes to topic list if given. */
- private def getConsumer(props: Properties, topics: Option[List[String]] = None) = {
- val keyDeserializer = new ByteArrayDeserializer
- val valueDeserializer = new ByteArrayDeserializer
- val consumer = new KafkaConsumer(props, keyDeserializer, valueDeserializer)
- topics foreach { consumer.subscribe(_) }
- consumer
- }
-
- private val consumer = getConsumer(getProps, Some(List(topic)))
-}
diff --git a/common/scala/src/main/scala/whisk/connector/kafka/KafkaMessagingProvider.scala b/common/scala/src/main/scala/whisk/connector/kafka/KafkaMessagingProvider.scala
index 7fd82c507f..1682eca13f 100644
--- a/common/scala/src/main/scala/whisk/connector/kafka/KafkaMessagingProvider.scala
+++ b/common/scala/src/main/scala/whisk/connector/kafka/KafkaMessagingProvider.scala
@@ -20,25 +20,19 @@ package whisk.connector.kafka
import java.util.Properties
import java.util.concurrent.ExecutionException
-import scala.concurrent.ExecutionContext
-import scala.concurrent.duration.FiniteDuration
-import scala.collection.JavaConverters._
-
-import org.apache.kafka.clients.admin.AdminClientConfig
-import org.apache.kafka.clients.admin.AdminClient
-import org.apache.kafka.clients.admin.NewTopic
+import akka.actor.ActorSystem
+import akka.stream.UniqueKillSwitch
+import akka.stream.scaladsl.Source
+import org.apache.kafka.clients.admin.{AdminClient, AdminClientConfig, NewTopic}
import org.apache.kafka.common.errors.TopicExistsException
-
+import pureconfig._
import whisk.common.Logging
import whisk.core.ConfigKeys
-import whisk.core.WhiskConfig
-import whisk.core.connector.MessageConsumer
-import whisk.core.connector.MessageProducer
-import whisk.core.connector.MessagingProvider
+import whisk.core.connector.{MessageProducer, MessagingProvider}
-import pureconfig._
+import scala.collection.JavaConverters._
-case class KafkaConfig(replicationFactor: Short)
+case class KafkaConfig(hosts: String, replicationFactor: Short)
case class TopicConfig(segmentBytes: Long, retentionBytes: Long, retentionMs: Long) {
def toMap: Map[String, String] = {
@@ -53,18 +47,19 @@ case class TopicConfig(segmentBytes: Long, retentionBytes: Long, retentionMs: Lo
* A Kafka based implementation of MessagingProvider
*/
object KafkaMessagingProvider extends MessagingProvider {
- def getConsumer(config: WhiskConfig, groupId: String, topic: String, maxPeek: Int, maxPollInterval: FiniteDuration)(
- implicit logging: Logging): MessageConsumer =
- new KafkaConsumerConnector(config.kafkaHosts, groupId, topic, maxPeek, maxPollInterval = maxPollInterval)
+ def getConsumer(group: String, topic: String, maxBatchSize: Int)(
+ implicit actorSystem: ActorSystem): Source[String, UniqueKillSwitch] = {
+ OwKafkaConsumer.bufferedSource(group, topic, maxBatchSize)
+ }
- def getProducer(config: WhiskConfig, ec: ExecutionContext)(implicit logging: Logging): MessageProducer =
- new KafkaProducerConnector(config.kafkaHosts, ec)
+ def getProducer()(implicit actorSystem: ActorSystem, logging: Logging): MessageProducer =
+ new KafkaProducerConnector()
- def ensureTopic(config: WhiskConfig, topic: String, topicConfig: String)(implicit logging: Logging): Boolean = {
+ def ensureTopic(topic: String, topicConfig: String)(implicit logging: Logging): Boolean = {
val kc = loadConfigOrThrow[KafkaConfig](ConfigKeys.kafka)
val tc = loadConfigOrThrow[TopicConfig](ConfigKeys.kafkaTopics + s".$topicConfig")
val props = new Properties
- props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.kafkaHosts)
+ props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kc.hosts)
val client = AdminClient.create(props)
val numPartitions = 1
val nt = new NewTopic(topic, numPartitions, kc.replicationFactor).configs(tc.toMap.asJava)
diff --git a/common/scala/src/main/scala/whisk/connector/kafka/KafkaProducerConnector.scala b/common/scala/src/main/scala/whisk/connector/kafka/KafkaProducerConnector.scala
index 18b38249bd..6a1a2939da 100644
--- a/common/scala/src/main/scala/whisk/connector/kafka/KafkaProducerConnector.scala
+++ b/common/scala/src/main/scala/whisk/connector/kafka/KafkaProducerConnector.scala
@@ -17,84 +17,53 @@
package whisk.connector.kafka
-import java.util.Properties
-
-import scala.concurrent.ExecutionContext
-import scala.concurrent.Future
-import scala.concurrent.Promise
-import scala.util.Failure
-import scala.util.Success
-import org.apache.kafka.clients.producer.Callback
-import org.apache.kafka.clients.producer.KafkaProducer
-import org.apache.kafka.clients.producer.ProducerConfig
-import org.apache.kafka.clients.producer.ProducerRecord
-import org.apache.kafka.clients.producer.RecordMetadata
-import org.apache.kafka.common.errors.NotLeaderForPartitionException
+import akka.actor.ActorSystem
+import akka.kafka.{ProducerMessage, ProducerSettings}
+import akka.stream.scaladsl.{Sink, Source, SourceQueueWithComplete}
+import akka.stream.{ActorMaterializer, OverflowStrategy}
+import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata}
import org.apache.kafka.common.serialization.StringSerializer
-import whisk.common.Counter
-import whisk.common.Logging
-import whisk.core.connector.Message
-import whisk.core.connector.MessageProducer
-import whisk.core.entity.UUIDs
+import whisk.common.{Logging, TransactionId}
+import whisk.core.connector.{Message, MessageProducer}
+
+import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.util.{Failure, Success}
+
+class KafkaProducerConnector()(implicit actorSystem: ActorSystem, logging: Logging) extends MessageProducer {
-class KafkaProducerConnector(kafkahosts: String,
- implicit val executionContext: ExecutionContext,
- id: String = UUIDs.randomUUID().toString)(implicit logging: Logging)
- extends MessageProducer {
+ val producerSettings = ProducerSettings(actorSystem, new StringSerializer, new StringSerializer)
- override def sentCount() = sentCounter.cur
+ private implicit val executionContext: ExecutionContext = actorSystem.dispatchers.lookup(producerSettings.dispatcher)
+ private implicit val materializer: ActorMaterializer = ActorMaterializer()
+
+ private val kafkaProducer: SourceQueueWithComplete[(ProducerRecord[String, String], Promise[RecordMetadata])] =
+ Source
+ .queue[(ProducerRecord[String, String], Promise[RecordMetadata])](Int.MaxValue, OverflowStrategy.dropNew)
+ .map { case (msg, prom) => ProducerMessage.Message(msg, prom) }
+ .via(OwKafkaProducer.flow(producerSettings))
+ .map(result => result.message.passThrough.success(result.metadata))
+ .to(Sink.ignore)
+ .run()
/** Sends msg to topic. This is an asynchronous operation. */
override def send(topic: String, msg: Message, retry: Int = 2): Future[RecordMetadata] = {
- implicit val transid = msg.transid
+ implicit val transid: TransactionId = msg.transid
val record = new ProducerRecord[String, String](topic, "messages", msg.serialize)
logging.debug(this, s"sending to topic '$topic' msg '$msg'")
val produced = Promise[RecordMetadata]()
- producer.send(record, new Callback {
- override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
- if (exception == null) produced.success(metadata)
- else produced.failure(exception)
- }
- })
+ // TODO handle all cases
+ kafkaProducer.offer((record, produced))
produced.future.andThen {
case Success(status) =>
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}")
- } recoverWith {
- case t: NotLeaderForPartitionException =>
- if (retry > 0) {
- logging.error(this, s"NotLeaderForPartitionException is retryable, remain $retry retry")
- Thread.sleep(100)
- send(topic, msg, retry - 1)
- } else produced.future
}
}
/** Closes producer. */
- override def close() = {
- logging.info(this, "closing producer")
- producer.close()
- }
-
- private val sentCounter = new Counter()
-
- private def getProps: Properties = {
- val props = new Properties
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkahosts)
- props.put(ProducerConfig.ACKS_CONFIG, 1.toString)
- props
- }
-
- private def getProducer(props: Properties): KafkaProducer[String, String] = {
- val keySerializer = new StringSerializer
- val valueSerializer = new StringSerializer
- new KafkaProducer(props, keySerializer, valueSerializer)
- }
-
- private val producer = getProducer(getProps)
+ override def close(): Unit = kafkaProducer.complete()
}
diff --git a/common/scala/src/main/scala/whisk/connector/kafka/OwKafkaConsumer.scala b/common/scala/src/main/scala/whisk/connector/kafka/OwKafkaConsumer.scala
new file mode 100644
index 0000000000..574c7eb73a
--- /dev/null
+++ b/common/scala/src/main/scala/whisk/connector/kafka/OwKafkaConsumer.scala
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.connector.kafka
+
+import akka.Done
+import akka.actor.ActorSystem
+import akka.kafka.ConsumerMessage.CommittableOffsetBatch
+import akka.kafka.scaladsl.Consumer
+import akka.kafka.{ConsumerSettings, Subscriptions}
+import akka.stream.scaladsl.{Keep, RestartSource, Source}
+import akka.stream.{KillSwitches, UniqueKillSwitch}
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.common.errors.RetriableException
+import org.apache.kafka.common.serialization.{ByteArrayDeserializer, StringDeserializer}
+
+import scala.collection.immutable.Queue
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration._
+
+object OwKafkaConsumer {
+
+ /**
+ * Creates a source of messages read from kafka, deserialized as Strings.
+ *
+ * @param group groupId to use to connect to kafka
+ * @param topic topic to subscribe to
+ * @param maxBufferSize the maximum size of messages buffered
+ * @return a source of messages from kafka
+ */
+ def bufferedSource(group: String, topic: String, maxBufferSize: Int)(
+ implicit actorSystem: ActorSystem): Source[String, UniqueKillSwitch] = {
+
+ batchedSouce(group, topic, maxBufferSize).mapConcat(identity)
+ }
+
+ /**
+ * Creates a source of messages read from kafka in batches, deserialized as Strings.
+ *
+ * @param group groupId to use to connect to kafka
+ * @param topic topic to subscribe to
+ * @param maxBatchSize the maximum batch size to read
+ * @return a source of batched messages from kafka
+ */
+ def batchedSouce(group: String, topic: String, maxBatchSize: Int)(
+ implicit actorSystem: ActorSystem): Source[Queue[String], UniqueKillSwitch] = {
+ val settings = ConsumerSettings(actorSystem, new ByteArrayDeserializer, new StringDeserializer)
+ .withGroupId(group)
+ .withProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxBatchSize.toString)
+
+ implicit val ec: ExecutionContext = actorSystem.dispatchers.lookup(settings.dispatcher)
+
+ RestartSource
+ .withBackoff(minBackoff = 500.milliseconds, maxBackoff = 30.seconds, randomFactor = 0.1) { () =>
+ Consumer
+ .committableSource(settings, Subscriptions.topics(topic))
+ .batch(maxBatchSize, Queue(_))(_ :+ _)
+ .mapAsync(4) { msgs =>
+ val batch =
+ msgs.foldLeft(CommittableOffsetBatch.empty)((batch, msg) => batch.updated(msg.committableOffset))
+
+ commitBatch(batch).map(_ => msgs.map(_.record.value))
+ }
+ }
+ .viaMat(KillSwitches.single)(Keep.right)
+ }
+
+ /**
+ * Commits a batch of messages and retries retriable exceptions.
+ *
+ * @param batch the batch to commit
+ * @return a Future that completes once the batch has been committed
+ */
+ private def commitBatch(batch: CommittableOffsetBatch)(implicit ec: ExecutionContext): Future[Done] = {
+ batch.commitScaladsl().recoverWith {
+ case e: RetriableException =>
+ // TODO backoff strategy for the retry
+ commitBatch(batch)
+ }
+ }
+
+}
diff --git a/common/scala/src/main/scala/whisk/connector/kafka/OwKafkaProducer.scala b/common/scala/src/main/scala/whisk/connector/kafka/OwKafkaProducer.scala
new file mode 100644
index 0000000000..e60767f211
--- /dev/null
+++ b/common/scala/src/main/scala/whisk/connector/kafka/OwKafkaProducer.scala
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.connector.kafka
+
+import akka.pattern.after
+import akka.NotUsed
+import akka.actor.ActorSystem
+import akka.kafka.ProducerMessage.{Message, Result}
+import akka.kafka.ProducerSettings
+import akka.stream.scaladsl.Flow
+import org.apache.kafka.clients.producer.{Callback, Producer, ProducerRecord, RecordMetadata}
+import org.apache.kafka.common.errors.RetriableException
+
+import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.concurrent.duration._
+
+/**
+ * Resilient Kafka producing flows. Modeled after the ideas in:
+ *
+ * - https://github.com/akka/reactive-kafka/issues/250
+ * - https://github.com/akka/reactive-kafka/issues/344
+ *
+ * We (@markusthoemmes/@cbickel) plan to contribute those to akka-streams-kafka once hardened.
+ */
+object OwKafkaProducer {
+
+ /**
+ * Produces a message in kafka.
+ *
+ * @param producer the KafkaProducer to use to produce the message
+ * @param record the record to produce
+ * @return a Future that completes with the RecordMetadata once the message is produced successfully
+ */
+ def produce[K, V](producer: Producer[K, V], record: ProducerRecord[K, V]): Future[RecordMetadata] = {
+ val r = Promise[RecordMetadata]
+ producer.send(
+ record,
+ new Callback {
+ override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = {
+ if (exception == null) {
+ r.success(metadata)
+ } else {
+ r.failure(exception)
+ }
+ }
+ })
+
+ r.future
+ }
+
+ /**
+ * Produces a message in kafka, retrying on retriable exceptions.
+ *
+ * @param producer the KafkaProducer to use to produce the message
+ * @param record the record to produce
+ * @return a Future that completes with the RecordMetadata once the message is produced successfully
+ */
+ def retryingProduce[K, V](producer: Producer[K, V], record: ProducerRecord[K, V])(
+ implicit ec: ExecutionContext,
+ system: ActorSystem): Future[RecordMetadata] = {
+
+ produce(producer, record).recoverWith {
+ case e: RetriableException =>
+ // TODO backoff strategy for the retry
+ after(50.milliseconds, system.scheduler)(retryingProduce(producer, record))
+ case t =>
+ Future.failed(t)
+ }
+ }
+
+ def flow[K, V, PassThrough](settings: ProducerSettings[K, V])(
+ implicit as: ActorSystem): Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed] = {
+ flow(settings, settings.createKafkaProducer())
+ }
+
+ def flow[K, V, PassThrough](settings: ProducerSettings[K, V], producer: Producer[K, V])(
+ implicit system: ActorSystem): Flow[Message[K, V, PassThrough], Result[K, V, PassThrough], NotUsed] = {
+ implicit val ec: ExecutionContext = system.dispatchers.lookup(settings.dispatcher)
+
+ Flow[Message[K, V, PassThrough]]
+ .mapAsync(settings.parallelism) { msg =>
+ retryingProduce(producer, msg.record).map(meta => Result(meta, msg))
+ }
+ }
+}
diff --git a/common/scala/src/main/scala/whisk/core/WhiskConfig.scala b/common/scala/src/main/scala/whisk/core/WhiskConfig.scala
index f0db605e94..58b3bbc2e4 100644
--- a/common/scala/src/main/scala/whisk/core/WhiskConfig.scala
+++ b/common/scala/src/main/scala/whisk/core/WhiskConfig.scala
@@ -75,7 +75,6 @@ class WhiskConfig(requiredProperties: Map[String, String],
val controllerInstances = this(WhiskConfig.controllerInstances)
val edgeHost = this(WhiskConfig.edgeHostName) + ":" + this(WhiskConfig.edgeHostApiPort)
- val kafkaHosts = this(WhiskConfig.kafkaHostList)
val edgeHostName = this(WhiskConfig.edgeHostName)
@@ -212,7 +211,6 @@ object WhiskConfig {
val loadbalancerInvokerBusyThreshold = "loadbalancer.invokerBusyThreshold"
- val kafkaHostList = "kafka.hosts"
val zookeeperHostList = "zookeeper.hosts"
private val edgeHostApiPort = "edge.host.apiport"
@@ -222,7 +220,6 @@ object WhiskConfig {
val edgeHost = Map(edgeHostName -> null, edgeHostApiPort -> null)
val invokerHosts = Map(invokerHostsList -> null)
- val kafkaHosts = Map(kafkaHostList -> null)
val zookeeperHosts = Map(zookeeperHostList -> null)
val runtimesManifest = "runtimes.manifest"
diff --git a/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala b/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
deleted file mode 100644
index 51977677ab..0000000000
--- a/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
+++ /dev/null
@@ -1,247 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package whisk.core.connector
-
-import scala.annotation.tailrec
-import scala.collection.immutable
-import scala.concurrent.Future
-import scala.concurrent.blocking
-import scala.concurrent.duration._
-import scala.util.Failure
-import org.apache.kafka.clients.consumer.CommitFailedException
-import akka.actor.FSM
-import akka.pattern.pipe
-import whisk.common.Logging
-import whisk.common.TransactionId
-
-trait MessageConsumer {
-
- /** The maximum number of messages peeked (i.e., max number of messages retrieved during a long poll). */
- val maxPeek: Int
-
- /**
- * Gets messages via a long poll. May or may not remove messages
- * from the message connector. Use commit() to ensure messages are
- * removed from the connector.
- *
- * @param duration for the long poll
- * @return iterable collection (topic, partition, offset, bytes)
- */
- def peek(duration: Duration): Iterable[(String, Int, Long, Array[Byte])]
-
- /**
- * Commits offsets from last peek operation to ensure they are removed
- * from the connector.
- */
- def commit(): Unit
-
- /** Closes consumer. */
- def close(): Unit
-
-}
-
-object MessageFeed {
- protected sealed trait FeedState
- protected[connector] case object Idle extends FeedState
- protected[connector] case object FillingPipeline extends FeedState
- protected[connector] case object DrainingPipeline extends FeedState
-
- protected sealed trait FeedData
- private case object NoData extends FeedData
-
- /** Indicates the consumer is ready to accept messages from the message bus for processing. */
- object Ready
-
- /** Steady state message, indicates capacity in downstream process to receive more messages. */
- object Processed
-
- /** Indicates the fill operation has completed. */
- private case class FillCompleted(messages: Seq[(String, Int, Long, Array[Byte])])
-}
-
-/**
- * This actor polls the message bus for new messages and dispatches them to the given
- * handler. The actor tracks the number of messages dispatched and will not dispatch new
- * messages until some number of them are acknowledged.
- *
- * This is used by the invoker to pull messages from the message bus and apply back pressure
- * when the invoker does not have resources to complete processing messages (i.e., no containers
- * are available to run new actions). It is also used in the load balancer to consume active
- * ack messages.
- * When the invoker releases resources (by reclaiming containers) it will send a message
- * to this actor which will then attempt to fill the pipeline with new messages.
- *
- * The actor tries to fill the pipeline with additional messages while the number
- * of outstanding requests is below the pipeline fill threshold.
- */
-@throws[IllegalArgumentException]
-class MessageFeed(description: String,
- logging: Logging,
- consumer: MessageConsumer,
- maximumHandlerCapacity: Int,
- longPollDuration: FiniteDuration,
- handler: Array[Byte] => Future[Unit],
- autoStart: Boolean = true,
- logHandoff: Boolean = true)
- extends FSM[MessageFeed.FeedState, MessageFeed.FeedData] {
- import MessageFeed._
-
- // double-buffer to make up for message bus read overhead
- val maxPipelineDepth = maximumHandlerCapacity * 2
- private val pipelineFillThreshold = maxPipelineDepth - consumer.maxPeek
-
- require(
- consumer.maxPeek <= maxPipelineDepth,
- "consumer may not yield more messages per peek than permitted by max depth")
-
- // Immutable Queue
- // although on the surface it seems to make sense to use an immutable variable with a mutable Queue,
- // Akka Actor state defies the usual "prefer immutable" guideline in Scala, esp. w/ Collections.
- // If, for some reason, this Queue was mutable and is accidentally leaked in say an Akka message,
- // another Actor or recipient would be able to mutate the internal state of this Actor.
- // Best practice dictates a mutable variable pointing at an immutable collection for this reason
- private var outstandingMessages = immutable.Queue.empty[(String, Int, Long, Array[Byte])]
- private var handlerCapacity = maximumHandlerCapacity
-
- private implicit val tid = TransactionId.dispatcher
-
- logging.info(
- this,
- s"handler capacity = $maximumHandlerCapacity, pipeline fill at = $pipelineFillThreshold, pipeline depth = $maxPipelineDepth")
-
- when(Idle) {
- case Event(Ready, _) =>
- fillPipeline()
- goto(FillingPipeline)
-
- case _ => stay
- }
-
- // wait for fill to complete, and keep filling if there is
- // capacity otherwise wait to drain
- when(FillingPipeline) {
- case Event(Processed, _) =>
- updateHandlerCapacity()
- sendOutstandingMessages()
- stay
-
- case Event(FillCompleted(messages), _) =>
- outstandingMessages = outstandingMessages ++ messages
- sendOutstandingMessages()
-
- if (shouldFillQueue()) {
- fillPipeline()
- stay
- } else {
- goto(DrainingPipeline)
- }
-
- case _ => stay
- }
-
- when(DrainingPipeline) {
- case Event(Processed, _) =>
- updateHandlerCapacity()
- sendOutstandingMessages()
- if (shouldFillQueue()) {
- fillPipeline()
- goto(FillingPipeline)
- } else stay
-
- case _ => stay
- }
-
- onTransition { case _ -> Idle => if (autoStart) self ! Ready }
- startWith(Idle, MessageFeed.NoData)
- initialize()
-
- private implicit val ec = context.system.dispatcher
-
- private def fillPipeline(): Unit = {
- if (outstandingMessages.size <= pipelineFillThreshold) {
- Future {
- blocking {
- // Grab next batch of messages and commit offsets immediately
- // essentially marking the activation as having satisfied "at most once"
- // semantics (this is the point at which the activation is considered started).
- // If the commit fails, then messages peeked are peeked again on the next poll.
- // While the commit is synchronous and will block until it completes, at steady
- // state with enough buffering (i.e., maxPipelineDepth > maxPeek), the latency
- // of the commit should be masked.
- val records = consumer.peek(longPollDuration)
- consumer.commit()
- FillCompleted(records.toSeq)
- }
- }.andThen {
- case Failure(e: CommitFailedException) =>
- logging.error(this, s"failed to commit $description consumer offset: $e")
- case Failure(e: Throwable) => logging.error(this, s"exception while pulling new $description records: $e")
- }
- .recover {
- case _ => FillCompleted(Seq.empty)
- }
- .pipeTo(self)
- } else {
- logging.error(this, s"dropping fill request until $description feed is drained")
- }
- }
-
- /** Send as many messages as possible to the handler. */
- @tailrec
- private def sendOutstandingMessages(): Unit = {
- val occupancy = outstandingMessages.size
- if (occupancy > 0 && handlerCapacity > 0) {
- // Easiest way with an immutable queue to cleanly dequeue
- // Head is the first elemeent of the queue, desugared w/ an assignment pattern
- // Tail is everything but the first element, thus mutating the collection variable
- val (topic, partition, offset, bytes) = outstandingMessages.head
- outstandingMessages = outstandingMessages.tail
-
- if (logHandoff) logging.info(this, s"processing $topic[$partition][$offset] ($occupancy/$handlerCapacity)")
- handler(bytes)
- handlerCapacity -= 1
-
- sendOutstandingMessages()
- }
- }
-
- private def shouldFillQueue(): Boolean = {
- val occupancy = outstandingMessages.size
- if (occupancy <= pipelineFillThreshold) {
- logging.debug(
- this,
- s"$description pipeline has capacity: $occupancy <= $pipelineFillThreshold ($handlerCapacity)")
- true
- } else {
- logging.debug(this, s"$description pipeline must drain: $occupancy > $pipelineFillThreshold")
- false
- }
- }
-
- private def updateHandlerCapacity(): Int = {
- logging.debug(self, s"$description received processed msg, current capacity = $handlerCapacity")
-
- if (handlerCapacity < maximumHandlerCapacity) {
- handlerCapacity += 1
- handlerCapacity
- } else {
- if (handlerCapacity > maximumHandlerCapacity) logging.error(self, s"$description capacity already at max")
- maximumHandlerCapacity
- }
- }
-}
diff --git a/common/scala/src/main/scala/whisk/core/connector/MessageProducer.scala b/common/scala/src/main/scala/whisk/core/connector/MessageProducer.scala
index de37c33917..48f5bb92ef 100644
--- a/common/scala/src/main/scala/whisk/core/connector/MessageProducer.scala
+++ b/common/scala/src/main/scala/whisk/core/connector/MessageProducer.scala
@@ -23,9 +23,6 @@ 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]
diff --git a/common/scala/src/main/scala/whisk/core/connector/MessagingProvider.scala b/common/scala/src/main/scala/whisk/core/connector/MessagingProvider.scala
index 1737624611..f8366ceb60 100644
--- a/common/scala/src/main/scala/whisk/core/connector/MessagingProvider.scala
+++ b/common/scala/src/main/scala/whisk/core/connector/MessagingProvider.scala
@@ -17,22 +17,20 @@
package whisk.core.connector
-import scala.concurrent.ExecutionContext
-import scala.concurrent.duration.DurationInt
-import scala.concurrent.duration.FiniteDuration
+import akka.actor.ActorSystem
+import akka.stream.UniqueKillSwitch
+import akka.stream.scaladsl.Source
import whisk.common.Logging
-import whisk.core.WhiskConfig
import whisk.spi.Spi
/**
* An Spi for providing Messaging implementations.
*/
trait MessagingProvider extends Spi {
- def getConsumer(config: WhiskConfig,
- groupId: String,
- topic: String,
- maxPeek: Int = Int.MaxValue,
- maxPollInterval: FiniteDuration = 5.minutes)(implicit logging: Logging): MessageConsumer
- def getProducer(config: WhiskConfig, ec: ExecutionContext)(implicit logging: Logging): MessageProducer
- def ensureTopic(config: WhiskConfig, topic: String, topicConfig: String)(implicit logging: Logging): Boolean
+ def getConsumer(group: String, topic: String, maxBatchSize: Int)(
+ implicit actorSystem: ActorSystem): Source[String, UniqueKillSwitch]
+
+ def getProducer()(implicit actorSystem: ActorSystem, logging: Logging): MessageProducer
+
+ def ensureTopic(topic: String, topicConfig: String)(implicit logging: Logging): Boolean
}
diff --git a/common/scala/src/main/scala/whisk/core/database/RemoteCacheInvalidation.scala b/common/scala/src/main/scala/whisk/core/database/RemoteCacheInvalidation.scala
index b615708e4f..58fddcc28a 100644
--- a/common/scala/src/main/scala/whisk/core/database/RemoteCacheInvalidation.scala
+++ b/common/scala/src/main/scala/whisk/core/database/RemoteCacheInvalidation.scala
@@ -17,31 +17,18 @@
package whisk.core.database
-import java.nio.charset.StandardCharsets
-
-import scala.concurrent.Future
-import scala.concurrent.duration.DurationInt
-import scala.util.Failure
-import scala.util.Success
-import scala.util.Try
-
import akka.actor.ActorSystem
-import akka.actor.Props
+import akka.stream.ActorMaterializer
import spray.json._
import whisk.common.Logging
import whisk.core.WhiskConfig
-import whisk.core.connector.Message
-import whisk.core.connector.MessageFeed
-import whisk.core.connector.MessagingProvider
-import whisk.core.entity.CacheKey
-import whisk.core.entity.InstanceId
-import whisk.core.entity.WhiskAction
-import whisk.core.entity.WhiskActionMetaData
-import whisk.core.entity.WhiskPackage
-import whisk.core.entity.WhiskRule
-import whisk.core.entity.WhiskTrigger
+import whisk.core.connector.{Message, MessagingProvider}
+import whisk.core.entity._
import whisk.spi.SpiLoader
+import scala.concurrent.Future
+import scala.util.{Failure, Success, Try}
+
case class CacheInvalidationMessage(key: CacheKey, instanceId: String) extends Message {
override def serialize = CacheInvalidationMessage.serdes.write(this).compactPrint
}
@@ -54,35 +41,26 @@ object CacheInvalidationMessage extends DefaultJsonProtocol {
class RemoteCacheInvalidation(config: WhiskConfig, component: String, instance: InstanceId)(implicit logging: Logging,
as: ActorSystem) {
+ implicit private val materializer = ActorMaterializer()
implicit private val ec = as.dispatcher
private val topic = "cacheInvalidation"
private val instanceId = s"$component${instance.toInt}"
private val msgProvider = SpiLoader.get[MessagingProvider]
- private val cacheInvalidationConsumer = msgProvider.getConsumer(config, s"$topic$instanceId", topic, maxPeek = 128)
- private val cacheInvalidationProducer = msgProvider.getProducer(config, ec)
+
+ msgProvider.getConsumer(s"$topic$instanceId", topic, 128).runForeach(removeFromLocalCache)
+
+ private val cacheInvalidationProducer = msgProvider.getProducer()
def notifyOtherInstancesAboutInvalidation(key: CacheKey): Future[Unit] = {
cacheInvalidationProducer.send(topic, CacheInvalidationMessage(key, instanceId)).map(_ => Unit)
}
- private val invalidationFeed = as.actorOf(Props {
- new MessageFeed(
- "cacheInvalidation",
- logging,
- cacheInvalidationConsumer,
- cacheInvalidationConsumer.maxPeek,
- 1.second,
- removeFromLocalCache)
- })
-
def invalidateWhiskActionMetaData(key: CacheKey) =
WhiskActionMetaData.removeId(key)
- private def removeFromLocalCache(bytes: Array[Byte]): Future[Unit] = Future {
- val raw = new String(bytes, StandardCharsets.UTF_8)
-
+ private def removeFromLocalCache(raw: String): Unit = {
CacheInvalidationMessage.parse(raw) match {
case Success(msg: CacheInvalidationMessage) => {
if (msg.instanceId != instanceId) {
@@ -95,6 +73,5 @@ class RemoteCacheInvalidation(config: WhiskConfig, component: String, instance:
}
case Failure(t) => logging.error(this, s"failed processing message: $raw with $t")
}
- invalidationFeed ! MessageFeed.Processed
}
}
diff --git a/core/controller/src/main/scala/whisk/core/controller/Controller.scala b/core/controller/src/main/scala/whisk/core/controller/Controller.scala
index 08b8f1d544..c62351851f 100644
--- a/core/controller/src/main/scala/whisk/core/controller/Controller.scala
+++ b/core/controller/src/main/scala/whisk/core/controller/Controller.scala
@@ -212,13 +212,13 @@ object Controller {
}
val msgProvider = SpiLoader.get[MessagingProvider]
- if (!msgProvider.ensureTopic(config, topic = "completed" + instance, topicConfig = "completed")) {
+ if (!msgProvider.ensureTopic(topic = "completed" + instance, topicConfig = "completed")) {
abort(s"failure during msgProvider.ensureTopic for topic completed$instance")
}
- if (!msgProvider.ensureTopic(config, topic = "health", topicConfig = "health")) {
+ if (!msgProvider.ensureTopic(topic = "health", topicConfig = "health")) {
abort(s"failure during msgProvider.ensureTopic for topic health")
}
- if (!msgProvider.ensureTopic(config, topic = "cacheInvalidation", topicConfig = "cache-invalidation")) {
+ if (!msgProvider.ensureTopic(topic = "cacheInvalidation", topicConfig = "cache-invalidation")) {
abort(s"failure during msgProvider.ensureTopic for topic cacheInvalidation")
}
diff --git a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
index cd80a8ee12..8f1f9c7731 100644
--- a/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
+++ b/core/controller/src/main/scala/whisk/core/loadBalancer/InvokerSupervision.scala
@@ -17,33 +17,24 @@
package whisk.core.loadBalancer
-import java.nio.charset.StandardCharsets
-
-import scala.collection.immutable
-import scala.concurrent.Future
-import scala.concurrent.duration._
-import scala.util.Failure
-import scala.util.Success
-import org.apache.kafka.clients.producer.RecordMetadata
-import akka.actor.Actor
-import akka.actor.ActorRef
-import akka.actor.ActorRefFactory
-import akka.actor.FSM
-import akka.actor.FSM.CurrentState
-import akka.actor.FSM.SubscribeTransitionCallBack
-import akka.actor.FSM.Transition
-import akka.actor.Props
+import akka.actor.FSM.{CurrentState, SubscribeTransitionCallBack, Transition}
+import akka.actor.{Actor, ActorRef, ActorRefFactory, FSM, Props}
import akka.pattern.pipe
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.Source
import akka.util.Timeout
-import whisk.common.AkkaLogging
-import whisk.common.LoggingMarkers
-import whisk.common.RingBuffer
-import whisk.common.TransactionId
+import org.apache.kafka.clients.producer.RecordMetadata
+import whisk.common.{AkkaLogging, LoggingMarkers, RingBuffer, TransactionId}
import whisk.core.connector._
import whisk.core.entitlement.Privilege
import whisk.core.entity.ActivationId.ActivationIdGenerator
import whisk.core.entity._
+import scala.collection.immutable
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import scala.util.{Failure, Success}
+
// Received events
case object GetStatus
@@ -74,9 +65,11 @@ final case class InvokerInfo(buffer: RingBuffer[Boolean])
*/
class InvokerPool(childFactory: (ActorRefFactory, InstanceId) => ActorRef,
sendActivationToInvoker: (ActivationMessage, InstanceId) => Future[RecordMetadata],
- pingConsumer: MessageConsumer)
+ pingConsumer: Source[String, _])
extends Actor {
+ implicit val materializer = ActorMaterializer()
+
implicit val transid = TransactionId.invokerHealth
implicit val logging = new AkkaLogging(context.system.log)
implicit val timeout = Timeout(5.seconds)
@@ -124,26 +117,14 @@ class InvokerPool(childFactory: (ActorRefFactory, InstanceId) => ActorRef,
/** Receive Ping messages from invokers. */
val pingPollDuration = 1.second
- val invokerPingFeed = context.system.actorOf(Props {
- new MessageFeed(
- "ping",
- logging,
- pingConsumer,
- pingConsumer.maxPeek,
- pingPollDuration,
- processInvokerPing,
- logHandoff = false)
- })
-
- def processInvokerPing(bytes: Array[Byte]): Future[Unit] = Future {
- val raw = new String(bytes, StandardCharsets.UTF_8)
+
+ pingConsumer.runForeach(processInvokerPing)
+
+ def processInvokerPing(raw: String): Unit = {
PingMessage.parse(raw) match {
case Success(p: PingMessage) =>
self ! p
- invokerPingFeed ! MessageFeed.Processed
-
case Failure(t) =>
- invokerPingFeed ! MessageFeed.Processed
logging.error(this, s"failed processing message: $raw with $t")
}
}
@@ -171,7 +152,7 @@ class InvokerPool(childFactory: (ActorRefFactory, InstanceId) => ActorRef,
object InvokerPool {
def props(f: (ActorRefFactory, InstanceId) => ActorRef,
p: (ActivationMessage, InstanceId) => Future[RecordMetadata],
- pc: MessageConsumer) = {
+ pc: Source[String, _]) = {
Props(new InvokerPool(f, p, pc))
}
diff --git a/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
index 89ba90045f..71c122a0fa 100644
--- a/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
+++ b/core/controller/src/main/scala/whisk/core/loadBalancer/LoadBalancerService.scala
@@ -17,33 +17,18 @@
package whisk.core.loadBalancer
-import java.nio.charset.StandardCharsets
-
-import scala.annotation.tailrec
-import scala.concurrent.Await
-import scala.concurrent.ExecutionContext
-import scala.concurrent.Future
-import scala.concurrent.Promise
-import scala.concurrent.duration._
-import scala.util.Failure
-import scala.util.Success
-import org.apache.kafka.clients.producer.RecordMetadata
-import akka.actor.ActorRefFactory
-import akka.actor.ActorSystem
-import akka.actor.Props
+import akka.actor.{ActorRefFactory, ActorSystem}
import akka.cluster.Cluster
-import akka.util.Timeout
import akka.pattern.ask
-import whisk.common.Logging
-import whisk.common.LoggingMarkers
-import whisk.common.TransactionId
import whisk.core.ConfigKeys
+import akka.stream.ActorMaterializer
+import akka.util.Timeout
+import org.apache.kafka.clients.producer.RecordMetadata
+import pureconfig._
+import whisk.common.{Logging, LoggingMarkers, TransactionId}
import whisk.core.WhiskConfig
import whisk.core.WhiskConfig._
-import whisk.core.connector.{ActivationMessage, CompletionMessage}
-import whisk.core.connector.MessageFeed
-import whisk.core.connector.MessageProducer
-import whisk.core.connector.MessagingProvider
+import whisk.core.connector.{ActivationMessage, CompletionMessage, MessageProducer, MessagingProvider}
import whisk.core.database.NoDocumentException
import whisk.core.entity._
import whisk.core.entity.{ActivationId, WhiskActivation}
@@ -55,7 +40,11 @@ import whisk.core.entity.UUID
import whisk.core.entity.WhiskAction
import whisk.core.entity.types.EntityStore
import whisk.spi.SpiLoader
-import pureconfig._
+
+import scala.annotation.tailrec
+import scala.concurrent.{Await, ExecutionContext, Future, Promise}
+import scala.concurrent.duration._
+import scala.util.{Failure, Success}
case class LoadbalancerConfig(blackboxFraction: Double, invokerBusyThreshold: Int)
@@ -90,6 +79,7 @@ class LoadBalancerService(config: WhiskConfig, instance: InstanceId, entityStore
implicit val actorSystem: ActorSystem,
logging: Logging)
extends LoadBalancer {
+ implicit val materializer = ActorMaterializer()
private val lbConfig = loadConfigOrThrow[LoadbalancerConfig](ConfigKeys.loadbalancer)
@@ -229,7 +219,7 @@ class LoadBalancerService(config: WhiskConfig, instance: InstanceId, entityStore
/** Gets a producer which can publish messages to the kafka bus. */
private val messagingProvider = SpiLoader.get[MessagingProvider]
- private val messageProducer = messagingProvider.getProducer(config, executionContext)
+ private val messageProducer = messagingProvider.getProducer()
private def sendActivationToInvoker(producer: MessageProducer,
msg: ActivationMessage,
@@ -265,7 +255,7 @@ class LoadBalancerService(config: WhiskConfig, instance: InstanceId, entityStore
val maxPingsPerPoll = 128
val pingConsumer =
- messagingProvider.getConsumer(config, s"health${instance.toInt}", "health", maxPeek = maxPingsPerPoll)
+ messagingProvider.getConsumer(s"health${instance.toInt}", "health", maxPingsPerPoll)
val invokerFactory = (f: ActorRefFactory, invokerInstance: InstanceId) =>
f.actorOf(InvokerActor.props(invokerInstance, instance))
@@ -280,26 +270,16 @@ class LoadBalancerService(config: WhiskConfig, instance: InstanceId, entityStore
val maxActiveAcksPerPoll = 128
val activeAckPollDuration = 1.second
private val activeAckConsumer =
- messagingProvider.getConsumer(config, "completions", s"completed${instance.toInt}", maxPeek = maxActiveAcksPerPoll)
- val activationFeed = actorSystem.actorOf(Props {
- new MessageFeed(
- "activeack",
- logging,
- activeAckConsumer,
- maxActiveAcksPerPoll,
- activeAckPollDuration,
- processActiveAck)
- })
-
- def processActiveAck(bytes: Array[Byte]): Future[Unit] = Future {
- val raw = new String(bytes, StandardCharsets.UTF_8)
+ messagingProvider.getConsumer("completions", s"completed${instance.toInt}", maxActiveAcksPerPoll)
+
+ activeAckConsumer.runForeach(processActiveAck)
+
+ def processActiveAck(raw: String): Future[Unit] = Future {
CompletionMessage.parse(raw) match {
case Success(m: CompletionMessage) =>
processCompletion(m.response, m.transid, forced = false, invoker = m.invoker)
- activationFeed ! MessageFeed.Processed
case Failure(t) =>
- activationFeed ! MessageFeed.Processed
logging.error(this, s"failed processing message: $raw with $t")
}
}
@@ -354,8 +334,7 @@ class LoadBalancerService(config: WhiskConfig, instance: InstanceId, entityStore
object LoadBalancerService {
def requiredProperties =
- kafkaHosts ++
- Map(controllerLocalBookkeeping -> null, controllerSeedNodes -> null)
+ Map(controllerLocalBookkeeping -> null, controllerSeedNodes -> null)
/** Memoizes the result of `f` for later use. */
def memoize[I, O](f: I => O): I => O = new scala.collection.mutable.HashMap[I, O]() {
diff --git a/core/invoker/build.gradle b/core/invoker/build.gradle
index 4a9ceff902..ef7827b704 100644
--- a/core/invoker/build.gradle
+++ b/core/invoker/build.gradle
@@ -18,8 +18,4 @@ dependencies {
compile 'org.apache.zookeeper:zookeeper:3.4.11'
}
-tasks.withType(ScalaCompile) {
- scalaCompileOptions.additionalParameters = gradle.scala.compileFlags
-}
-
mainClassName = "whisk.core.invoker.Invoker"
diff --git a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
index 61172ddd7a..5afe5179eb 100644
--- a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
+++ b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
@@ -31,12 +31,17 @@ import whisk.core.entity.CodeExec
import whisk.core.entity.EntityName
import whisk.core.entity.ExecutableWhiskAction
import whisk.core.entity.size._
-import whisk.core.connector.MessageFeed
sealed trait WorkerState
case object Busy extends WorkerState
case object Free extends WorkerState
+object FlowControl {
+ case object Initialize
+ case object Ignore
+ case object Processed
+}
+
case class WorkerData(data: ContainerData, state: WorkerState)
/**
@@ -57,13 +62,11 @@ case class WorkerData(data: ContainerData, state: WorkerState)
* @param childFactory method to create new container proxy actor
* @param maxActiveContainers maximum amount of containers doing work
* @param maxPoolSize maximum size of containers allowed in the pool
- * @param feed actor to request more work from
* @param prewarmConfig optional settings for container prewarming
*/
class ContainerPool(childFactory: ActorRefFactory => ActorRef,
maxActiveContainers: Int,
maxPoolSize: Int,
- feed: ActorRef,
prewarmConfig: Option[PrewarmingConfig] = None)
extends Actor {
implicit val logging = new AkkaLogging(context.system.log)
@@ -72,6 +75,8 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
var busyPool = immutable.Map.empty[ActorRef, ContainerData]
var prewarmedPool = immutable.Map.empty[ActorRef, ContainerData]
+ var feed = ActorRef.noSender
+
prewarmConfig.foreach { config =>
logging.info(this, s"pre-warming ${config.count} ${config.exec.kind} containers")(TransactionId.invokerWarmup)
(1 to config.count).foreach { _ =>
@@ -80,6 +85,10 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
}
def receive: Receive = {
+ case FlowControl.Initialize =>
+ feed = sender()
+ (0 to maxActiveContainers).foreach(_ => feed ! FlowControl.Processed)
+
// A job to run on a container
//
// Run messages are received either via the feed or from child containers which cannot process
@@ -126,7 +135,7 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
freePool = freePool + (sender() -> data)
busyPool.get(sender()).foreach { _ =>
busyPool = busyPool - sender()
- feed ! MessageFeed.Processed
+ feed ! FlowControl.Processed
}
// Container is prewarmed and ready to take work
@@ -138,8 +147,7 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
freePool = freePool - sender()
busyPool.get(sender()).foreach { _ =>
busyPool = busyPool - sender()
- // container was busy, so there is capacity to accept another job request
- feed ! MessageFeed.Processed
+ feed ! FlowControl.Processed
}
// This message is received for one of these reasons:
@@ -249,9 +257,8 @@ object ContainerPool {
def props(factory: ActorRefFactory => ActorRef,
maxActive: Int,
size: Int,
- feed: ActorRef,
prewarmConfig: Option[PrewarmingConfig] = None) =
- Props(new ContainerPool(factory, maxActive, size, feed, prewarmConfig))
+ Props(new ContainerPool(factory, maxActive, size, prewarmConfig))
}
/** Contains settings needed to perform container prewarming */
diff --git a/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala b/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
index 648e61e072..36bfa526f0 100644
--- a/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
+++ b/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
@@ -60,7 +60,6 @@ object Invoker {
ExecManifest.requiredProperties ++
WhiskEntityStore.requiredProperties ++
WhiskActivationStore.requiredProperties ++
- kafkaHosts ++
zookeeperHosts ++
wskApiHost ++ Map(
dockerImageTag -> "latest",
@@ -182,10 +181,10 @@ object Invoker {
val invokerInstance = InstanceId(assignedInvokerId)
val msgProvider = SpiLoader.get[MessagingProvider]
- if (!msgProvider.ensureTopic(config, topic = "invoker" + assignedInvokerId, topicConfig = "invoker")) {
+ if (!msgProvider.ensureTopic(topic = "invoker" + assignedInvokerId, topicConfig = "invoker")) {
abort(s"failure during msgProvider.ensureTopic for topic invoker$assignedInvokerId")
}
- val producer = msgProvider.getProducer(config, ec)
+ val producer = msgProvider.getProducer()
val invoker = try {
new InvokerReactive(config, invokerInstance, producer)
} catch {
diff --git a/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala b/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
index c885c8581d..2d7a6e5731 100644
--- a/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
+++ b/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
@@ -17,35 +17,18 @@
package whisk.core.invoker
-import java.nio.charset.StandardCharsets
import java.time.Instant
-import scala.concurrent.Future
-import scala.concurrent.duration._
-import scala.util.Failure
-import scala.util.Success
-
+import akka.actor.{ActorRefFactory, ActorSystem}
+import akka.stream.scaladsl.Sink
+import akka.stream.{ActorAttributes, ActorMaterializer, Supervision}
import org.apache.kafka.common.errors.RecordTooLargeException
-
-import akka.actor.ActorRefFactory
-import akka.actor.ActorSystem
-import akka.actor.Props
-import akka.stream.ActorMaterializer
+import spray.json.DefaultJsonProtocol._
import spray.json._
-import whisk.common.Logging
-import whisk.common.LoggingMarkers
-import whisk.common.TransactionId
+import whisk.common.{Logging, LoggingMarkers, TransactionId}
import whisk.core.WhiskConfig
-import whisk.core.connector.ActivationMessage
-import whisk.core.connector.CompletionMessage
-import whisk.core.connector.MessageFeed
-import whisk.core.connector.MessageProducer
-import whisk.core.connector.MessagingProvider
-import whisk.core.containerpool.ContainerFactoryProvider
-import whisk.core.containerpool.ContainerPool
-import whisk.core.containerpool.ContainerProxy
-import whisk.core.containerpool.PrewarmingConfig
-import whisk.core.containerpool.Run
+import whisk.core.connector.{ActivationMessage, CompletionMessage, MessageProducer, MessagingProvider}
+import whisk.core.containerpool._
import whisk.core.containerpool.logging.LogStoreProvider
import whisk.core.database._
import whisk.core.entity._
@@ -53,6 +36,9 @@ import whisk.core.entity.size._
import whisk.http.Messages
import whisk.spi.SpiLoader
+import scala.concurrent.Future
+import scala.util.{Failure, Success}
+
class InvokerReactive(config: WhiskConfig, instance: InstanceId, producer: MessageProducer)(
implicit actorSystem: ActorSystem,
logging: Logging) {
@@ -95,16 +81,6 @@ class InvokerReactive(config: WhiskConfig, instance: InstanceId, producer: Messa
val topic = s"invoker${instance.toInt}"
val maximumContainers = config.invokerNumCore.toInt * config.invokerCoreShare.toInt
val msgProvider = SpiLoader.get[MessagingProvider]
- val consumer = msgProvider.getConsumer(
- config,
- topic,
- topic,
- maximumContainers,
- maxPollInterval = TimeLimit.MAX_DURATION + 1.minute)
-
- val activationFeed = actorSystem.actorOf(Props {
- new MessageFeed("activation", logging, consumer, maximumContainers, 500.milliseconds, processActivationMessage)
- })
/** Sends an active-ack. */
val ack = (tid: TransactionId,
@@ -153,94 +129,84 @@ class InvokerReactive(config: WhiskConfig, instance: InstanceId, producer: Messa
.get
val pool = actorSystem.actorOf(
- ContainerPool.props(
- childFactory,
- maximumContainers,
- maximumContainers,
- activationFeed,
- Some(PrewarmingConfig(2, prewarmExec, 256.MB))))
-
- /** Is called when an ActivationMessage is read from Kafka */
- def processActivationMessage(bytes: Array[Byte]): Future[Unit] = {
- Future(ActivationMessage.parse(new String(bytes, StandardCharsets.UTF_8)))
- .flatMap(Future.fromTry(_))
- .filter(_.action.version.isDefined)
- .flatMap { msg =>
- implicit val transid = msg.transid
-
- val start = transid.started(this, LoggingMarkers.INVOKER_ACTIVATION)
- val namespace = msg.action.path
- val name = msg.action.name
- val actionid = FullyQualifiedEntityName(namespace, name).toDocId.asDocInfo(msg.revision)
- val subject = msg.user.subject
-
- logging.info(this, s"${actionid.id} $subject ${msg.activationId}")
-
- // caching is enabled since actions have revision id and an updated
- // action will not hit in the cache due to change in the revision id;
- // if the doc revision is missing, then bypass cache
- if (actionid.rev == DocRevision.empty) {
- logging.warn(this, s"revision was not provided for ${actionid.id}")
- }
-
- WhiskAction
- .get(entityStore, actionid.id, actionid.rev, fromCache = actionid.rev != DocRevision.empty)
- .flatMap { action =>
- action.toExecutableWhiskAction match {
- case Some(executable) =>
- pool ! Run(executable, msg)
- Future.successful(())
- case None =>
- logging.error(this, s"non-executable action reached the invoker ${action.fullyQualifiedName(false)}")
- Future.failed(new IllegalStateException("non-executable action reached the invoker"))
- }
- }
- .recoverWith {
- case t =>
- // If the action cannot be found, the user has concurrently deleted it,
- // making this an application error. All other errors are considered system
- // errors and should cause the invoker to be considered unhealthy.
- val response = t match {
- case _: NoDocumentException =>
- ActivationResponse.applicationError(Messages.actionRemovedWhileInvoking)
- case _: DocumentTypeMismatchException | _: DocumentUnreadable =>
- ActivationResponse.whiskError(Messages.actionMismatchWhileInvoking)
- case _ =>
- ActivationResponse.whiskError(Messages.actionFetchErrorWhileInvoking)
- }
- val now = Instant.now
- val causedBy = if (msg.causedBySequence) {
- Some(Parameters(WhiskActivation.causedByAnnotation, JsString(Exec.SEQUENCE)))
- } else None
- val activation = WhiskActivation(
- activationId = msg.activationId,
- namespace = msg.user.namespace.toPath,
- subject = msg.user.subject,
- cause = msg.cause,
- name = msg.action.name,
- version = msg.action.version.getOrElse(SemVer()),
- start = now,
- end = now,
- duration = Some(0),
- response = response,
- annotations = {
- Parameters(WhiskActivation.pathAnnotation, JsString(msg.action.asString)) ++ causedBy
- })
-
- activationFeed ! MessageFeed.Processed
- ack(msg.transid, activation, msg.blocking, msg.rootControllerIndex)
- store(msg.transid, activation)
- Future.successful(())
- }
- }
- .recoverWith {
- case t =>
- // Iff everything above failed, we have a terminal error at hand. Either the message failed
- // to deserialize, or something threw an error where it is not expected to throw.
- activationFeed ! MessageFeed.Processed
- logging.error(this, s"terminal failure while processing message: $t")
- Future.successful(())
+ ContainerPool
+ .props(childFactory, maximumContainers, maximumContainers, Some(PrewarmingConfig(2, prewarmExec, 256.MB))))
+
+ msgProvider
+ .getConsumer(topic, topic, maximumContainers * 2)
+ .map(msg => ActivationMessage.parse(msg))
+ .filter {
+ case Success(_) => true
+ case Failure(t) =>
+ logging.error(this, s"terminal failure while processing message: $t")
+ false
+ }
+ .collect {
+ case Success(msg) => msg
+ }
+ .filter(_.action.version.isDefined)
+ .mapAsyncUnordered(8) { msg =>
+ implicit val transid = msg.transid
+
+ val start = transid.started(this, LoggingMarkers.INVOKER_ACTIVATION)
+ val namespace = msg.action.path
+ val name = msg.action.name
+ val actionid = FullyQualifiedEntityName(namespace, name).toDocId.asDocInfo(msg.revision)
+ val subject = msg.user.subject
+
+ logging.info(this, s"${actionid.id} $subject ${msg.activationId}")
+
+ // caching is enabled since actions have revision id and an updated
+ // action will not hit in the cache due to change in the revision id;
+ // if the doc revision is missing, then bypass cache
+ if (actionid.rev == DocRevision.empty) {
+ logging.warn(this, s"revision was not provided for ${actionid.id}")
}
- }
+ WhiskAction
+ .get(entityStore, actionid.id, actionid.rev, fromCache = actionid.rev != DocRevision.empty)
+ .flatMap { action =>
+ action.toExecutableWhiskAction match {
+ case Some(executable) =>
+ Future.successful(Run(executable, msg))
+ case None =>
+ logging.error(this, s"non-executable action reached the invoker ${action.fullyQualifiedName(false)}")
+ Future.failed(new IllegalStateException("non-executable action reached the invoker"))
+ }
+ }
+ .recoverWith {
+ case t =>
+ // If the action cannot be found, the user has concurrently deleted it,
+ // making this an application error. All other errors are considered system
+ // errors and should cause the invoker to be considered unhealthy.
+ val response = t match {
+ case _: NoDocumentException => ActivationResponse.applicationError(Messages.actionRemovedWhileInvoking)
+ case _ => ActivationResponse.whiskError(Messages.actionMismatchWhileInvoking)
+ }
+ val now = Instant.now
+ val causedBy = if (msg.causedBySequence) Parameters("causedBy", "sequence".toJson) else Parameters()
+ val activation = WhiskActivation(
+ activationId = msg.activationId,
+ namespace = msg.user.namespace.toPath,
+ subject = msg.user.subject,
+ cause = msg.cause,
+ name = msg.action.name,
+ version = msg.action.version.getOrElse(SemVer()),
+ start = now,
+ end = now,
+ duration = Some(0),
+ response = response,
+ annotations = {
+ Parameters("path", msg.action.toString.toJson) ++ causedBy
+ })
+
+ //activationFeed ! MessageFeed.Processed
+ ack(msg.transid, activation, msg.blocking, msg.rootControllerIndex)
+ store(msg.transid, activation)
+ Future.failed(new IllegalStateException("concurrent action deletion"))
+ }
+ }
+ // Per docs, Resume will just drop failed message
+ .withAttributes(ActorAttributes.supervisionStrategy(_ => Supervision.Resume))
+ .runWith(Sink.actorRefWithAck(pool, FlowControl.Initialize, FlowControl.Processed, FlowControl.Ignore))
}
diff --git a/tests/src/test/resources/application.conf b/tests/src/test/resources/application.conf.j2
similarity index 86%
rename from tests/src/test/resources/application.conf
rename to tests/src/test/resources/application.conf.j2
index 039f2f4eb7..12c3d547bd 100644
--- a/tests/src/test/resources/application.conf
+++ b/tests/src/test/resources/application.conf.j2
@@ -8,3 +8,5 @@ whisk.spi {
akka.http.client.idle-timeout = 90 s
akka.http.host-connection-pool.idle-timeout = 90 s
akka.http.host-connection-pool.client.idle-timeout = 90 s
+
+whisk.kafka.hosts = "{{ kafka_connect_string }}"
\ No newline at end of file
diff --git a/tests/src/test/scala/services/KafkaConnectorTests.scala b/tests/src/test/scala/services/KafkaConnectorTests.scala
index 6001c9cd58..2745a331ef 100644
--- a/tests/src/test/scala/services/KafkaConnectorTests.scala
+++ b/tests/src/test/scala/services/KafkaConnectorTests.scala
@@ -18,50 +18,46 @@
package services
import java.io.File
-import java.util.Calendar
+import java.time.Instant
-import scala.concurrent.Await
-import scala.concurrent.duration.{DurationInt, FiniteDuration}
-import scala.language.postfixOps
-import scala.util.Try
-import org.apache.kafka.clients.consumer.CommitFailedException
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{Keep, Sink}
+import akka.testkit.TestProbe
+import common.{StreamLogging, TestUtils, WhiskProperties, WskActorSystem}
import org.junit.runner.RunWith
-import org.scalatest.BeforeAndAfterAll
-import org.scalatest.FlatSpec
-import org.scalatest.Matchers
import org.scalatest.junit.JUnitRunner
-import common.{StreamLogging, TestUtils, WhiskProperties, WskActorSystem}
+import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
+import pureconfig.loadConfigOrThrow
import whisk.common.TransactionId
-import whisk.connector.kafka.KafkaConsumerConnector
-import whisk.connector.kafka.KafkaProducerConnector
+import whisk.connector.kafka.{KafkaConfig, KafkaMessagingProvider}
import whisk.core.WhiskConfig
import whisk.core.connector.Message
-import whisk.utils.ExecutionContextFactory
-import whisk.utils.retry
+import whisk.utils.{retry, ExecutionContextFactory}
+
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
+import scala.language.postfixOps
+import scala.util.Try
@RunWith(classOf[JUnitRunner])
class KafkaConnectorTests extends FlatSpec with Matchers with WskActorSystem with BeforeAndAfterAll with StreamLogging {
implicit val transid = TransactionId.testing
implicit val ec = ExecutionContextFactory.makeCachedThreadPoolExecutionContext()
- val config = new WhiskConfig(WhiskConfig.kafkaHosts)
- assert(config.isValid)
+ implicit val materializer = ActorMaterializer()
+
+ val kafkaConfig = loadConfigOrThrow[KafkaConfig]("whisk.kafka")
+ val kafkaHosts = kafkaConfig.hosts.split(",")
val groupid = "kafkatest"
val topic = "Dinosaurs"
val sessionTimeout = 10 seconds
val maxPollInterval = 10 seconds
- val producer = new KafkaProducerConnector(config.kafkaHosts, ec)
- val consumer = new KafkaConsumerConnector(
- config.kafkaHosts,
- groupid,
- topic,
- sessionTimeout = sessionTimeout,
- maxPollInterval = maxPollInterval)
-
- override def afterAll() = {
+ val producer = KafkaMessagingProvider.getProducer()
+ val consumer = KafkaMessagingProvider.getConsumer(groupid, topic, 1)
+
+ override def afterAll() {
producer.close()
- consumer.close()
super.afterAll()
}
@@ -74,85 +70,60 @@ class KafkaConnectorTests extends FlatSpec with Matchers with WskActorSystem wit
TestUtils.runCmd(0, new File("."), cmd: _*)
}
- def sendAndReceiveMessage(message: Message,
- waitForSend: FiniteDuration,
- waitForReceive: FiniteDuration): Iterable[String] = {
- val start = java.lang.System.currentTimeMillis
- println(s"Send message to topic.\n")
- val sent = Await.result(producer.send(topic, message), waitForSend)
- println(s"Successfully sent message to topic: ${sent}\n")
- println(s"Receiving message from topic.\n")
- val received = consumer.peek(waitForReceive).map { case (_, _, _, msg) => new String(msg, "utf-8") }
- val end = java.lang.System.currentTimeMillis
- val elapsed = end - start
- println(s"Received ${received.size}. Took $elapsed msec: $received\n")
-
- received
- }
-
behavior of "Kafka connector"
it should "send and receive a kafka message which sets up the topic" in {
+ val probe = TestProbe()
+ val killSwitch = consumer.toMat(Sink.actorRef(probe.ref, ()))(Keep.left).run()
+
for (i <- 0 until 5) {
- val message = new Message { override val serialize = Calendar.getInstance().getTime().toString }
- val received = sendAndReceiveMessage(message, 20 seconds, 10 seconds)
- received.size should be >= 1
- received.last should be(message.serialize)
- consumer.commit()
- }
- }
+ val message = new Message { override val serialize = Instant.now.toString }
- it should "send and receive a kafka message even after session timeout" in {
- for (i <- 0 until 4) {
- val message = new Message { override val serialize = Calendar.getInstance().getTime().toString }
- val received = sendAndReceiveMessage(message, 1 seconds, 1 seconds)
-
- // only the last iteration will have an updated cursor
- // iteration 0: get whatever is on the topic (at least 1 but may be more if a previous test failed)
- // iteration 1: get iteration 0 records + 1 more (since we intentionally failed the commit on previous iteration)
- // iteration 2: get iteration 1 records + 1 more (since we intentionally failed the commit on previous iteration)
- // iteration 3: get exactly 1 records since iteration 2 should have forwarded the cursor
- if (i < 3) {
- received.size should be >= i + 1
- } else {
- received.size should be(1)
- }
- received.last should be(message.serialize)
-
- if (i < 2) {
- Thread.sleep((maxPollInterval + 1.second).toMillis)
- a[CommitFailedException] should be thrownBy {
- consumer.commit() // sleep should cause commit to fail
- }
- } else consumer.commit()
+ producer.send(topic, message)
+ probe.expectMsg(message.serialize)
}
+
+ killSwitch.shutdown()
}
it should "send and receive a kafka message even after shutdown one of instances" in {
- val kafkaHosts = config.kafkaHosts.split(",")
- if (kafkaHosts.length > 1) {
- for (i <- 0 until kafkaHosts.length) {
- val message = new Message { override val serialize = Calendar.getInstance().getTime().toString }
- val kafkaHost = kafkaHosts(i).split(":")(0)
- val startLog = s", started"
- val prevCount = startLog.r.findAllMatchIn(commandComponent(kafkaHost, "logs", s"kafka$i").stdout).length
-
- commandComponent(kafkaHost, "stop", s"kafka$i")
- var received = sendAndReceiveMessage(message, 30 seconds, 30 seconds)
- received.size should be(1)
- consumer.commit()
-
- commandComponent(kafkaHost, "start", s"kafka$i")
- retry({
- startLog.r
- .findAllMatchIn(commandComponent(kafkaHost, "logs", s"kafka$i").stdout)
- .length shouldBe prevCount + 1
- }, 20, Some(1.second)) // wait until kafka is up
-
- received = sendAndReceiveMessage(message, 30 seconds, 30 seconds)
- received.size should be(1)
- consumer.commit()
+
+ val probe = TestProbe()
+ val killSwitch = consumer.toMat(Sink.actorRef(probe.ref, ()))(Keep.left).run()
+
+ for (i <- 0 until kafkaHosts.length) {
+ def message = new Message {
+ override val serialize = Instant.now.toString
}
+
+ val kafkaHost = kafkaHosts(i).split(":")(0)
+ val startLog = ", started"
+ val prevCount = startLog.r.findAllMatchIn(commandComponent(kafkaHost, "logs", s"kafka$i").stdout).length
+
+ println(s"Stop kafka$i now")
+ commandComponent(kafkaHost, "stop", s"kafka$i")
+
+ // With only 1 Kafka, it will not be possible to send and receive messages
+ if (kafkaHosts.length > 1) {
+ val firstMessage = message
+ producer.send(topic, firstMessage)
+ probe.expectMsg(firstMessage.serialize)
+ }
+
+ println(s"Start kafka$i now")
+ commandComponent(kafkaHost, "start", s"kafka$i")
+ retry({
+ startLog.r
+ .findAllMatchIn(commandComponent(kafkaHost, "logs", s"kafka$i").stdout)
+ .length shouldBe prevCount + 1
+ }, 20, Some(1.second)) // wait until kafka is up
+ println(s"kafka$i started")
+
+ val secondMessage = message
+ Await.result(producer.send(topic, secondMessage), 1.minute)
+ probe.expectMsg(2.minutes, secondMessage.serialize)
+
+ killSwitch.shutdown()
}
}
}
diff --git a/tests/src/test/scala/whisk/core/connector/test/MessageFeedTests.scala b/tests/src/test/scala/whisk/core/connector/test/MessageFeedTests.scala
deleted file mode 100644
index b240eddf8a..0000000000
--- a/tests/src/test/scala/whisk/core/connector/test/MessageFeedTests.scala
+++ /dev/null
@@ -1,187 +0,0 @@
-/*
- * Copyright 2015-2016 IBM Corporation
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package whisk.core.connector.test
-
-import java.util.concurrent.atomic.AtomicInteger
-
-import scala.collection.mutable.Buffer
-import scala.concurrent.Future
-import scala.concurrent.duration._
-
-import org.junit.runner.RunWith
-import org.scalatest.BeforeAndAfterAll
-import org.scalatest.BeforeAndAfterEach
-import org.scalatest.FlatSpecLike
-import org.scalatest.Matchers
-import org.scalatest.junit.JUnitRunner
-
-import akka.actor.ActorRef
-import akka.actor.ActorSystem
-import akka.actor.FSM
-import akka.actor.FSM.CurrentState
-import akka.actor.FSM.SubscribeTransitionCallBack
-import akka.actor.FSM.Transition
-import akka.actor.PoisonPill
-import akka.actor.Props
-import akka.testkit.TestKit
-import common.StreamLogging
-import whisk.core.connector._
-import whisk.core.connector.MessageFeed._
-import whisk.utils.retry
-
-@RunWith(classOf[JUnitRunner])
-class MessageFeedTests
- extends FlatSpecLike
- with Matchers
- with BeforeAndAfterEach
- with BeforeAndAfterAll
- with StreamLogging {
-
- val system = ActorSystem("MessageFeedTestSystem")
- val actorsToDestroyAfterEach: Buffer[ActorRef] = Buffer.empty
-
- override def afterEach() = {
- actorsToDestroyAfterEach.foreach { _ ! PoisonPill }
- actorsToDestroyAfterEach.clear()
- }
-
- override def afterAll() = TestKit.shutdownActorSystem(system)
-
- case class Connector(autoStart: Boolean = true) extends TestKit(system) {
- val peekCount = new AtomicInteger()
-
- val consumer = new TestConnector("feedtest", 4, true) {
- override def peek(duration: Duration) = {
- peekCount.incrementAndGet()
- super.peek(duration)
- }
- }
-
- val sentCount = new AtomicInteger()
-
- def fill(n: Int) = {
- val msgs = (1 to n).map { _ =>
- new Message {
- override def serialize = {
- sentCount.incrementAndGet().toString
- }
- override def toString = {
- s"message${sentCount.get}"
- }
- }
- }
- consumer.send(msgs)
- }
-
- val receivedCount = new AtomicInteger()
-
- def handler(bytes: Array[Byte]): Future[Unit] = {
- Future.successful(receivedCount.incrementAndGet())
- }
-
- val fsm = childActorOf(
- Props(new MessageFeed("test", logging, consumer, consumer.maxPeek, 200.milliseconds, handler, autoStart)))
-
- actorsToDestroyAfterEach += (fsm, testActor)
-
- def monitorTransitionsAndStart() = {
- fsm ! SubscribeTransitionCallBack(testActor)
- expectMsg(CurrentState(fsm, Idle))
- fsm ! Ready
- expectMsg(Transition(fsm, Idle, FillingPipeline))
- this
- }
- }
-
- def timeout(actor: ActorRef) = actor ! FSM.StateTimeout
-
- it should "wait for ready before accepting messages" in {
- val connector = Connector(autoStart = false)
- connector.fsm ! SubscribeTransitionCallBack(connector.testActor)
-
- // start idle
- connector.expectMsg(CurrentState(connector.fsm, Idle))
-
- // stay until received ready
- connector.fsm ! FSM.StateTimeout // should be ignored
- connector.fsm ! Processed // should be ignored
- Thread.sleep(500.milliseconds.toMillis)
- connector.peekCount.get shouldBe 0
-
- // start filling
- connector.fsm ! Ready
- connector.expectMsg(Transition(connector.fsm, Idle, FillingPipeline))
- retry(connector.peekCount.get should be > 0)
- }
-
- it should "auto start and start polling for messages" in {
- val connector = Connector(autoStart = true)
- // automatically start filling
- retry(connector.peekCount.get should be > 0, 5, Some(200.milliseconds))
- }
-
- it should "stop polling for messages when the pipeline is full" in {
- val connector = Connector(autoStart = false).monitorTransitionsAndStart()
- // push enough to cause pipeline to exceed fill mark
- connector.fill(connector.consumer.maxPeek * 2 + 1)
- retry(connector.peekCount.get should be > 0)
- retry(connector.receivedCount.get shouldBe connector.consumer.maxPeek, 10, Some(200.milliseconds))
-
- val peeks = connector.peekCount.get
- connector.expectMsg(Transition(connector.fsm, FillingPipeline, DrainingPipeline))
-
- connector.peekCount.get shouldBe peeks
- connector.expectNoMsg(500.milliseconds)
- }
-
- it should "transition from drain to fill mode" in {
- val connector = Connector(autoStart = false).monitorTransitionsAndStart()
- println(connector.fsm.toString())
- // push enough to cause pipeline to exceed fill mark
- val sendCount = connector.consumer.maxPeek * 2 + 2
- connector.fill(sendCount)
- retry(connector.peekCount.get should be > 0)
- retry(connector.receivedCount.get shouldBe connector.consumer.maxPeek, 10, Some(200.milliseconds))
-
- val peeks = connector.peekCount.get
- connector.expectMsg(Transition(connector.fsm, FillingPipeline, DrainingPipeline))
-
- // stay in drain mode, no more peeking
- timeout(connector.fsm) // should be ignored
- connector.expectNoMsg(500.milliseconds)
- connector.peekCount.get shouldBe peeks // no new reads
-
- // expecting overflow of 2 in the queue, which is true if all expected messages were sent
- retry(connector.sentCount.get shouldBe sendCount, 5, Some(200.milliseconds))
-
- // drain one, should stay in draining state
- connector.fsm ! Processed
- connector.expectNoMsg(500.milliseconds)
- connector.peekCount.get shouldBe peeks // no new reads
-
- // back to fill mode
- connector.fsm ! Processed
- connector.expectMsg(Transition(connector.fsm, DrainingPipeline, FillingPipeline))
- retry(connector.peekCount.get should be >= (peeks + 1))
-
- // should send back to drain mode
- connector.fill(1)
- connector.expectMsg(Transition(connector.fsm, FillingPipeline, DrainingPipeline))
-
- connector.expectNoMsg(500.milliseconds)
- }
-}
diff --git a/tests/src/test/scala/whisk/core/connector/test/TestConnector.scala b/tests/src/test/scala/whisk/core/connector/test/TestConnector.scala
deleted file mode 100644
index e3b9597a3f..0000000000
--- a/tests/src/test/scala/whisk/core/connector/test/TestConnector.scala
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package whisk.core.connector.test
-
-import java.util.ArrayList
-import java.util.concurrent.LinkedBlockingQueue
-
-import scala.concurrent.Future
-import scala.concurrent.duration.Duration
-import scala.collection.JavaConversions._
-
-import org.apache.kafka.clients.producer.RecordMetadata
-import org.apache.kafka.common.TopicPartition
-import common.StreamLogging
-
-import whisk.common.Counter
-import whisk.core.connector.Message
-import whisk.core.connector.MessageConsumer
-import whisk.core.connector.MessageProducer
-
-class TestConnector(topic: String, override val maxPeek: Int, allowMoreThanMax: Boolean)
- extends MessageConsumer
- with StreamLogging {
-
- override def peek(duration: Duration) = {
- val msgs = new ArrayList[Message]
- queue.synchronized {
- queue.drainTo(msgs, if (allowMoreThanMax) Int.MaxValue else maxPeek)
- msgs map { m =>
- offset += 1
- (topic, -1, offset, m.serialize.getBytes)
- }
- }
- }
-
- override def commit() = {
- if (throwCommitException) {
- throw new Exception("commit failed")
- } else {
- // nothing to do
- }
- }
-
- def occupancy = queue.size
-
- def send(msg: Message): Future[RecordMetadata] = {
- producer.send(topic, msg)
- }
-
- def send(msgs: Seq[Message]): Future[RecordMetadata] = {
- import scala.language.reflectiveCalls
- producer.sendBulk(topic, msgs)
- }
-
- def close() = {
- closed = true
- producer.close()
- }
-
- private val producer = new MessageProducer {
- def send(topic: String, msg: Message, retry: Int = 0): Future[RecordMetadata] = {
- queue.synchronized {
- if (queue.offer(msg)) {
- logging.info(this, s"put: $msg")
- Future.successful(new RecordMetadata(new TopicPartition(topic, 0), 0, queue.size, -1, Long.box(-1L), -1, -1))
- } else {
- logging.error(this, s"put failed: $msg")
- Future.failed(new IllegalStateException("failed to write msg"))
- }
- }
- }
-
- def sendBulk(topic: String, msgs: Seq[Message]): Future[RecordMetadata] = {
- queue.synchronized {
- if (queue.addAll(msgs)) {
- logging.info(this, s"put: ${msgs.length} messages")
- Future.successful(new RecordMetadata(new TopicPartition(topic, 0), 0, queue.size, -1, Long.box(-1L), -1, -1))
- } else {
- logging.error(this, s"put failed: ${msgs.length} messages")
- Future.failed(new IllegalStateException("failed to write msg"))
- }
- }
- }
-
- def close() = {}
- def sentCount() = counter.next()
- val counter = new Counter()
- }
-
- var throwCommitException = false
- private val queue = new LinkedBlockingQueue[Message]()
- @volatile private var closed = false
- private var offset = -1L
-}
diff --git a/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala b/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
index 0b531033d1..a9f41ea39a 100644
--- a/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
+++ b/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
@@ -42,7 +42,6 @@ import whisk.core.entity._
import whisk.core.entity.ExecManifest.RuntimeManifest
import whisk.core.entity.ExecManifest.ImageName
import whisk.core.entity.size._
-import whisk.core.connector.MessageFeed
/**
* Behavior tests for the ContainerPool
@@ -123,7 +122,7 @@ class ContainerPoolTests
it should "reuse a warm container" in within(timeout) {
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 2, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 2, 2))
pool ! runMessage
containers(0).expectMsg(runMessage)
@@ -137,7 +136,7 @@ class ContainerPoolTests
it should "reuse a warm container when action is the same even if revision changes" in within(timeout) {
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 2, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 2, 2))
pool ! runMessage
containers(0).expectMsg(runMessage)
@@ -152,7 +151,7 @@ class ContainerPoolTests
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 2, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 2, 2))
pool ! runMessage
containers(0).expectMsg(runMessage)
// Note that the container doesn't respond, thus it's not free to take work
@@ -166,11 +165,11 @@ class ContainerPoolTests
val feed = TestProbe()
// a pool with only 1 slot
- val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1))
pool ! runMessage
containers(0).expectMsg(runMessage)
containers(0).send(pool, NeedWork(warmedData()))
- feed.expectMsg(MessageFeed.Processed)
+ feed.expectMsg(FlowControl.Processed)
pool ! runMessageDifferentEverything
containers(0).expectMsg(Remove)
containers(1).expectMsg(runMessageDifferentEverything)
@@ -181,19 +180,19 @@ class ContainerPoolTests
val feed = TestProbe()
// a pool with only 1 active slot but 2 slots in total
- val pool = system.actorOf(ContainerPool.props(factory, 1, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 2))
// Run the first container
pool ! runMessage
containers(0).expectMsg(runMessage)
containers(0).send(pool, NeedWork(warmedData(lastUsed = Instant.EPOCH)))
- feed.expectMsg(MessageFeed.Processed)
+ feed.expectMsg(FlowControl.Processed)
// Run the second container, don't remove the first one
pool ! runMessageDifferentEverything
containers(1).expectMsg(runMessageDifferentEverything)
containers(1).send(pool, NeedWork(warmedData(lastUsed = Instant.now)))
- feed.expectMsg(MessageFeed.Processed)
+ feed.expectMsg(FlowControl.Processed)
pool ! runMessageDifferentNamespace
containers(2).expectMsg(runMessageDifferentNamespace)
@@ -207,11 +206,11 @@ class ContainerPoolTests
val feed = TestProbe()
// a pool with only 1 slot
- val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1))
pool ! runMessage
containers(0).expectMsg(runMessage)
containers(0).send(pool, NeedWork(warmedData()))
- feed.expectMsg(MessageFeed.Processed)
+ feed.expectMsg(FlowControl.Processed)
pool ! runMessageDifferentNamespace
containers(0).expectMsg(Remove)
containers(1).expectMsg(runMessageDifferentNamespace)
@@ -222,7 +221,7 @@ class ContainerPoolTests
val feed = TestProbe()
// a pool with only 1 slot
- val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1))
pool ! runMessage
containers(0).expectMsg(runMessage)
containers(0).send(pool, RescheduleJob) // emulate container failure ...
@@ -239,7 +238,7 @@ class ContainerPoolTests
val feed = TestProbe()
val pool =
- system.actorOf(ContainerPool.props(factory, 0, 0, feed.ref, Some(PrewarmingConfig(1, exec, memoryLimit))))
+ system.actorOf(ContainerPool.props(factory, 0, 0, Some(PrewarmingConfig(1, exec, memoryLimit))))
containers(0).expectMsg(Start(exec, memoryLimit))
}
@@ -248,7 +247,7 @@ class ContainerPoolTests
val feed = TestProbe()
val pool =
- system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref, Some(PrewarmingConfig(1, exec, memoryLimit))))
+ system.actorOf(ContainerPool.props(factory, 1, 1, Some(PrewarmingConfig(1, exec, memoryLimit))))
containers(0).expectMsg(Start(exec, memoryLimit))
containers(0).send(pool, NeedWork(preWarmedData(exec.kind)))
pool ! runMessage
@@ -261,8 +260,8 @@ class ContainerPoolTests
val alternativeExec = CodeExecAsString(RuntimeManifest("anotherKind", ImageName("testImage")), "testCode", None)
- val pool = system.actorOf(
- ContainerPool.props(factory, 1, 1, feed.ref, Some(PrewarmingConfig(1, alternativeExec, memoryLimit))))
+ val pool =
+ system.actorOf(ContainerPool.props(factory, 1, 1, Some(PrewarmingConfig(1, alternativeExec, memoryLimit))))
containers(0).expectMsg(Start(alternativeExec, memoryLimit)) // container0 was prewarmed
containers(0).send(pool, NeedWork(preWarmedData(alternativeExec.kind)))
pool ! runMessage
@@ -276,7 +275,7 @@ class ContainerPoolTests
val alternativeLimit = 128.MB
val pool =
- system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref, Some(PrewarmingConfig(1, exec, alternativeLimit))))
+ system.actorOf(ContainerPool.props(factory, 1, 1, Some(PrewarmingConfig(1, exec, alternativeLimit))))
containers(0).expectMsg(Start(exec, alternativeLimit)) // container0 was prewarmed
containers(0).send(pool, NeedWork(preWarmedData(exec.kind, alternativeLimit)))
pool ! runMessage
@@ -290,7 +289,7 @@ class ContainerPoolTests
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 2, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 2, 2))
// container0 is created and used
pool ! runMessage
diff --git a/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala b/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
index 8e7d2911cd..c662fadfe7 100644
--- a/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
+++ b/tests/src/test/scala/whisk/core/loadBalancer/test/InvokerSupervisionTests.scala
@@ -37,6 +37,7 @@ import akka.actor.FSM.CurrentState
import akka.actor.FSM.SubscribeTransitionCallBack
import akka.actor.FSM.Transition
import akka.pattern.ask
+import akka.stream.scaladsl.Source
import akka.testkit.ImplicitSender
import akka.testkit.TestFSMRef
import akka.testkit.TestKit
@@ -69,7 +70,6 @@ import whisk.core.loadBalancer.InvokerState
import whisk.core.loadBalancer.Offline
import whisk.core.loadBalancer.UnHealthy
import whisk.utils.retry
-import whisk.core.connector.test.TestConnector
import whisk.core.entitlement.Privilege
@RunWith(classOf[JUnitRunner])
@@ -104,7 +104,7 @@ class InvokerSupervisionTests
case (state, index) => (InstanceId(index), state)
}
- val pC = new TestConnector("pingFeedTtest", 4, false) {}
+ val pC = Source.empty[String]
behavior of "InvokerPool"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services