tests/src/test/scala/system/packages/MessageHubFeedTests.scala - openwhisk-package-kafka - Git at Google

 /*
  * 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 system.packages

 import system.utils.KafkaUtils

 import scala.concurrent.duration.DurationInt
 import scala.language.postfixOps

 import org.junit.runner.RunWith
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.FlatSpec
 import org.scalatest.Matchers
 import org.scalatest.Inside
 import org.scalatest.junit.JUnitRunner
 import org.apache.kafka.clients.producer.ProducerRecord

 import spray.json.DefaultJsonProtocol._
 import spray.json._

 import common.JsHelpers
 import common.TestUtils
 import common.TestHelpers
 import common.Wsk
 import common.WskActorSystem
 import common.WskProps
 import common.WskTestHelpers

 import ActionHelper._

 import java.util.Base64
 import java.nio.charset.StandardCharsets
 import java.time.{Clock, Instant}

 import whisk.utils.retry

 @RunWith(classOf[JUnitRunner])
 class MessageHubFeedTests
   extends FlatSpec
   with Matchers
   with Inside
   with WskActorSystem
   with BeforeAndAfterAll
   with TestHelpers
   with WskTestHelpers
   with JsHelpers {

   val topic = "test"
   val sessionTimeout = 10 seconds

   val messagingPackage = "/whisk.system/messaging"
   val messageHubFeed = "messageHubFeed"
   val messageHubProduce = "messageHubProduce"

   val consumerInitTime = 10000 // ms

   val kafkaUtils = new KafkaUtils

   val maxRetries = System.getProperty("max.retries", "60").toInt

   implicit val wskprops = WskProps()
   val wsk = new Wsk()
   val actionName = s"${messagingPackage}/${messageHubFeed}"

   val defaultAction = Some(TestUtils.getTestActionFilename("hello.js"))

   behavior of "Message Hub feed action"

   it should "reject invocation when topic argument is missing" in {
     val expectedOutput = JsObject(
       "error" -> JsString("You must supply a 'topic' parameter.")
     )

     runActionWithExpectedResult(actionName, "dat/missingTopic.json", expectedOutput, false)
   }

   it should "reject invocation when kafka_brokers_sasl argument is missing" in {
     val expectedOutput = JsObject(
       "error" -> JsString("You must supply a 'kafka_brokers_sasl' parameter.")
     )

     runActionWithExpectedResult(actionName, "dat/missingBrokers.json", expectedOutput, false)
   }

   it should "reject invocation when kafka_admin_url argument is missing" in {
     val expectedOutput = JsObject("error" -> JsString(
       "You must supply a 'kafka_admin_url' parameter.")
     )

     runActionWithExpectedResult(actionName, "dat/missingAdminURL.json", expectedOutput, false)
   }

   it should "reject invocation when user argument is missing" in {
     val expectedOutput = JsObject(
       "error" -> JsString("You must supply a 'user' parameter to authenticate with Message Hub.")
     )

     runActionWithExpectedResult(actionName, "dat/missingUser.json", expectedOutput, false)
   }

   it should "reject invocation when password argument is missing" in {
     val expectedOutput = JsObject(
       "error" -> JsString("You must supply a 'password' parameter to authenticate with Message Hub.")
     )

     runActionWithExpectedResult(actionName, "dat/missingPassword.json", expectedOutput, false)
   }

   it should "reject invocation when isJSONData and isBinaryValue are both enable" in {
     val expectedOutput = JsObject(
       "error" -> JsString("isJSONData and isBinaryValue cannot both be enabled."))

     runActionWithExpectedResult(actionName, "dat/multipleValueTypes.json", expectedOutput, false)
   }

   it should "fire a trigger when a binary message is posted to message hub" in withAssetCleaner(wskprops) {
     val currentTime = s"${System.currentTimeMillis}"

     (wp, assetHelper) =>
       val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
       println(s"Creating trigger ${triggerName}")

       createTrigger(assetHelper, triggerName, parameters = Map(
         "user" -> kafkaUtils.getAsJson("user"),
         "password" -> kafkaUtils.getAsJson("password"),
         "api_key" -> kafkaUtils.getAsJson("api_key"),
         "kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
         "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
         "topic" -> topic.toJson,
         "isBinaryKey" -> true.toJson,
         "isBinaryValue" -> true.toJson))

       val defaultActionName = s"helloKafka-${currentTime}"

       assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
         action.create(name, defaultAction)
       }
       assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
         rule.create(name, trigger = triggerName, action = defaultActionName)
       }

       // It takes a moment for the consumer to fully initialize.
       println("Giving the consumer a moment to get ready")
       Thread.sleep(consumerInitTime)

       // key to use for the produced message
       val key = "TheKey"
       val encodedCurrentTime = Base64.getEncoder.encodeToString(currentTime.getBytes(StandardCharsets.UTF_8))
       val encodedKey = Base64.getEncoder.encodeToString(key.getBytes(StandardCharsets.UTF_8))

       withActivation(wsk.activation, wsk.action.invoke(s"$messagingPackage/$messageHubProduce", Map(
         "user" -> kafkaUtils.getAsJson("user"),
         "password" -> kafkaUtils.getAsJson("password"),
         "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
         "topic" -> topic.toJson,
         "key" -> key.toJson,
         "value" -> currentTime.toJson))) {
         _.response.success shouldBe true
       }

       retry({
         println("Polling for activations")
         val activations = wsk.activation.pollFor(N = 1, Some(triggerName), retries = maxRetries)
         assert(activations.length == 1)

         val matchingActivations = for {
           id <- activations
           activation = wsk.activation.waitForActivation(id)
           if (activation.isRight && activation.right.get.fields.get("response").toString.contains(encodedCurrentTime))
         } yield activation.right.get

         assert(matchingActivations.length > 0)

         val activation = matchingActivations.head
         activation.getFieldPath("response", "success") shouldBe Some(true.toJson)

         // assert that there exists a message in the activation which has the expected keys and values
         val messages = KafkaUtils.messagesInActivation(activation, field = "value", value = encodedCurrentTime)
         assert(messages.length == 1)

         val message = messages.head
         message.getFieldPath("topic") shouldBe Some(topic.toJson)
         message.getFieldPath("key") shouldBe Some(encodedKey.toJson)
       }, N = 3)
   }

   it should "not fire a single trigger with an oversized payload" in withAssetCleaner(wskprops) {
     // payload size should be under the payload limit, but greater than 50% of the limit
     val testPayloadSize = 600000

     // create trigger
     val currentTime = s"${System.currentTimeMillis}"

     (wp, assetHelper) =>
       val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
       println(s"Creating trigger ${triggerName}")

       createTrigger(assetHelper, triggerName, parameters = Map(
         "user" -> kafkaUtils.getAsJson("user"),
         "password" -> kafkaUtils.getAsJson("password"),
         "api_key" -> kafkaUtils.getAsJson("api_key"),
         "kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
         "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
         "topic" -> topic.toJson,
         "isBinaryKey" -> false.toJson,
         "isBinaryValue" -> false.toJson))

       val defaultActionName = s"helloKafka-${currentTime}"

       assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
         action.create(name, defaultAction)
       }
       assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
         rule.create(name, trigger = triggerName, action = defaultActionName)
       }

       // It takes a moment for the consumer to fully initialize.
       println("Giving the consumer a moment to get ready")
       Thread.sleep(consumerInitTime)

       // Rapidly produce two messages whose size are each greater than half the allowed payload limit.
       // This should ensure that the feed fires these as two separate triggers.
       println("Rapidly producing two large messages")
       val producer = kafkaUtils.createProducer()
       val firstMessage = new ProducerRecord(topic, "key", generateMessage(s"first${currentTime}", testPayloadSize))
       val secondMessage = new ProducerRecord(topic, "key", generateMessage(s"second${currentTime}", testPayloadSize))
       producer.send(firstMessage)
       producer.send(secondMessage)
       producer.close()

       retry({
         // verify there are two trigger activations required to handle these messages
         println("Polling for activations")
         val activations = wsk.activation.pollFor(N = 2, Some(triggerName), retries = maxRetries)

         println("Verifying activation content")
         val matchingActivations = for {
           id <- activations
           activation = wsk.activation.waitForActivation(id)
           if (activation.isRight && (activation.right.get.fields.get("response").toString.contains(s"first${currentTime}") ||
             activation.right.get.fields.get("response").toString.contains(s"second${currentTime}")))
         } yield activation.right.get

         assert(matchingActivations.length == 2)
       }, N = 3)
   }

   it should "not fire a trigger for a single oversized message" in withAssetCleaner(wskprops) {
     // payload size should be over the payload limit
     val testPayloadSize = 2000000

     // create trigger
     val currentTime = s"${System.currentTimeMillis}"

     (wp, assetHelper) =>
       val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
       println(s"Creating trigger ${triggerName}")

       createTrigger(assetHelper, triggerName, parameters = Map(
         "user" -> kafkaUtils.getAsJson("user"),
         "password" -> kafkaUtils.getAsJson("password"),
         "api_key" -> kafkaUtils.getAsJson("api_key"),
         "kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
         "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
         "topic" -> topic.toJson,
         "isBinaryKey" -> false.toJson,
         "isBinaryValue" -> false.toJson))

       val defaultActionName = s"helloKafka-${currentTime}"

       assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
         action.create(name, defaultAction)
       }
       assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
         rule.create(name, trigger = triggerName, action = defaultActionName)
       }

       // It takes a moment for the consumer to fully initialize.
       println("Giving the consumer a moment to get ready")
       Thread.sleep(consumerInitTime)

       println("Producing an oversized message")
       val producer = kafkaUtils.createProducer()
       val bigMessage = new ProducerRecord(topic, "key", generateMessage(s"${currentTime}", testPayloadSize))
       producer.send(bigMessage)
       producer.close()

       retry({
         // verify there are no activations that match
         println("Polling for activations")
         val activations = wsk.activation.pollFor(N = 1, Some(triggerName), retries = maxRetries)

         println("Verifying activation content")
         val matchingActivations = for {
           id <- activations
           activation = wsk.activation.waitForActivation(id)
           if (activation.isRight && (activation.right.get.fields.get("response").toString.contains(s"first${currentTime}")))
         } yield activation.right.get

         assert(matchingActivations.length == 0)
       }, N = 3)
   }

   it should "reject trigger update without passing in any updatable parameters" in withAssetCleaner(wskprops) {
     val currentTime = s"${System.currentTimeMillis}"

     (wp, assetHelper) =>
       val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
       println(s"Creating trigger ${triggerName}")

       val username = kafkaUtils.getAsJson("user")
       val password = kafkaUtils.getAsJson("password")
       val admin_url = kafkaUtils.getAsJson("kafka_admin_url")
       val brokers = kafkaUtils.getAsJson("brokers")

       createTrigger(assetHelper, triggerName, parameters = Map(
         "user" -> username,
         "password" -> password,
         "api_key" -> kafkaUtils.getAsJson("api_key"),
         "kafka_admin_url" -> admin_url,
         "kafka_brokers_sasl" -> brokers,
         "topic" -> topic.toJson,
         "isBinaryKey" -> false.toJson,
         "isBinaryValue" -> false.toJson
       ))

       val run = wsk.action.invoke(actionName, parameters = Map(
         "triggerName" -> triggerName.toJson,
         "lifecycleEvent" -> "UPDATE".toJson,
         "authKey" -> wp.authKey.toJson
       ))

       withActivation(wsk.activation, run) {
         _.response.success shouldBe false
       }
   }

   it should "reject trigger update when both isJSONData and isBinaryValue are enabled" in withAssetCleaner(wskprops) {
     val currentTime = s"${System.currentTimeMillis}"

     (wp, assetHelper) =>
       val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
       println(s"Creating trigger $triggerName")

       val username = kafkaUtils.getAsJson("user")
       val password = kafkaUtils.getAsJson("password")
       val admin_url = kafkaUtils.getAsJson("kafka_admin_url")
       val brokers = kafkaUtils.getAsJson("brokers")

       createTrigger(assetHelper, triggerName, parameters = Map(
         "user" -> username,
         "password" -> password,
         "api_key" -> kafkaUtils.getAsJson("api_key"),
         "kafka_admin_url" -> admin_url,
         "kafka_brokers_sasl" -> brokers,
         "topic" -> topic.toJson,
         "isJSONData" -> true.toJson,
         "isBinaryKey" -> false.toJson,
         "isBinaryValue" -> false.toJson
       ))

       val run = wsk.action.invoke(actionName, parameters = Map(
         "triggerName" -> triggerName.toJson,
         "lifecycleEvent" -> "UPDATE".toJson,
         "authKey" -> wp.authKey.toJson,
         "isBinaryValue" -> true.toJson
       ))

       withActivation(wsk.activation, run) {
         _.response.success shouldBe false
       }
   }

   it should "fire a trigger when a message is posted to message hub before and after update" in withAssetCleaner(wskprops) {
     val currentTime = s"${System.currentTimeMillis}"

     (wp, assetHelper) =>
       val key = "TheKey"
       val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
       println(s"Creating trigger $triggerName")

       createTrigger(assetHelper, triggerName, parameters = Map(
         "user" -> kafkaUtils.getAsJson("user"),
         "password" -> kafkaUtils.getAsJson("password"),
         "api_key" -> kafkaUtils.getAsJson("api_key"),
         "kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
         "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
         "topic" -> topic.toJson
       ))

       val defaultActionName = s"helloKafka-${currentTime}"

       assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
         action.create(name, defaultAction)
       }
       assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
         rule.create(name, trigger = triggerName, action = defaultActionName)
       }

       println("Giving the consumer a moment to get ready")
       Thread.sleep(consumerInitTime)

       val first = Instant.now(Clock.systemUTC())

       println("Producing a message")
       withActivation(wsk.activation, wsk.action.invoke(s"$messagingPackage/$messageHubProduce", Map(
         "user" -> kafkaUtils.getAsJson("user"),
         "password" -> kafkaUtils.getAsJson("password"),
         "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
         "topic" -> topic.toJson,
         "key" -> key.toJson,
         "value" -> currentTime.toJson
       ))) {
         _.response.success shouldBe true
       }

       checkForActivations(triggerName, first, topic, key, currentTime)

       println("Updating trigger")

       val updateRunResult = wsk.action.invoke(actionName, parameters = Map(
         "triggerName" -> triggerName.toJson,
         "lifecycleEvent" -> "UPDATE".toJson,
         "authKey" -> wp.authKey.toJson,
         "isBinaryValue" -> true.toJson
       ))

       withActivation(wsk.activation, updateRunResult) {
         _.response.success shouldBe true
       }

       println("Giving the consumer a moment to get ready")
       Thread.sleep(consumerInitTime)

       val second = Instant.now(Clock.systemUTC())
       val encodedCurrentTime = Base64.getEncoder.encodeToString(currentTime.getBytes(StandardCharsets.UTF_8))

       println("Producing a message")
       withActivation(wsk.activation, wsk.action.invoke(s"$messagingPackage/$messageHubProduce", Map(
         "user" -> kafkaUtils.getAsJson("user"),
         "password" -> kafkaUtils.getAsJson("password"),
         "kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
         "topic" -> topic.toJson,
         "key" -> key.toJson,
         "value" -> currentTime.toJson
       ))) {
         _.response.success shouldBe true
       }

       checkForActivations(triggerName, second, topic, key, encodedCurrentTime)
   }

   def createTrigger(assetHelper: AssetCleaner, name: String, parameters: Map[String, spray.json.JsValue]) = {
     val feedCreationResult = assetHelper.withCleaner(wsk.trigger, name) {
       (trigger, _) =>
         trigger.create(name, feed = Some(s"$messagingPackage/$messageHubFeed"), parameters = parameters)
     }

     withActivation(wsk.activation, feedCreationResult, initialWait = 5 seconds, totalWait = 60 seconds) {
       activation =>
         // should be successful
         activation.response.success shouldBe true
     }
   }

   def checkForActivations(triggerName: String, since: Instant, topic: String, key: String, value: String) = {
     retry({
       println("Polling for activations")
       val activations = wsk.activation.pollFor(N = 1, Some(triggerName), since = Some(since), retries = maxRetries)
       assert(activations.length == 1)

       println("Validating content of activation(s)")
       val matchingActivations = for {
         id <- activations
         activation = wsk.activation.waitForActivation(id)
         if (activation.isRight && activation.right.get.fields.get("response").toString.contains(value))
       } yield activation.right.get

       assert(matchingActivations.length > 0)

       val activation = matchingActivations.head
       activation.getFieldPath("response", "success") shouldBe Some(true.toJson)

       // assert that there exists a message in the activation which has the expected keys and values
       val messages = KafkaUtils.messagesInActivation(activation, field = "value", value = value)
       assert(messages.length == 1)

       val message = messages.head
       message.getFieldPath("topic") shouldBe Some(topic.toJson)
       message.getFieldPath("key") shouldBe Some(key.toJson)
     }, N = 3)
   }

   def generateMessage(prefix: String, size: Int): String = {
     val longString = Array.fill[String](size)("0").mkString("")
     s"${prefix}${longString}"
   }
 }
	/*
	* 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 system.packages

	import system.utils.KafkaUtils

	import scala.concurrent.duration.DurationInt
	import scala.language.postfixOps

	import org.junit.runner.RunWith
	import org.scalatest.BeforeAndAfterAll
	import org.scalatest.FlatSpec
	import org.scalatest.Matchers
	import org.scalatest.Inside
	import org.scalatest.junit.JUnitRunner
	import org.apache.kafka.clients.producer.ProducerRecord

	import spray.json.DefaultJsonProtocol._
	import spray.json._

	import common.JsHelpers
	import common.TestUtils
	import common.TestHelpers
	import common.Wsk
	import common.WskActorSystem
	import common.WskProps
	import common.WskTestHelpers

	import ActionHelper._

	import java.util.Base64
	import java.nio.charset.StandardCharsets
	import java.time.{Clock, Instant}

	import whisk.utils.retry

	@RunWith(classOf[JUnitRunner])
	class MessageHubFeedTests
	extends FlatSpec
	with Matchers
	with Inside
	with WskActorSystem
	with BeforeAndAfterAll
	with TestHelpers
	with WskTestHelpers
	with JsHelpers {

	val topic = "test"
	val sessionTimeout = 10 seconds

	val messagingPackage = "/whisk.system/messaging"
	val messageHubFeed = "messageHubFeed"
	val messageHubProduce = "messageHubProduce"

	val consumerInitTime = 10000 // ms

	val kafkaUtils = new KafkaUtils

	val maxRetries = System.getProperty("max.retries", "60").toInt

	implicit val wskprops = WskProps()
	val wsk = new Wsk()
	val actionName = s"${messagingPackage}/${messageHubFeed}"

	val defaultAction = Some(TestUtils.getTestActionFilename("hello.js"))

	behavior of "Message Hub feed action"

	it should "reject invocation when topic argument is missing" in {
	val expectedOutput = JsObject(
	"error" -> JsString("You must supply a 'topic' parameter.")
	)

	runActionWithExpectedResult(actionName, "dat/missingTopic.json", expectedOutput, false)
	}

	it should "reject invocation when kafka_brokers_sasl argument is missing" in {
	val expectedOutput = JsObject(
	"error" -> JsString("You must supply a 'kafka_brokers_sasl' parameter.")
	)

	runActionWithExpectedResult(actionName, "dat/missingBrokers.json", expectedOutput, false)
	}

	it should "reject invocation when kafka_admin_url argument is missing" in {
	val expectedOutput = JsObject("error" -> JsString(
	"You must supply a 'kafka_admin_url' parameter.")
	)

	runActionWithExpectedResult(actionName, "dat/missingAdminURL.json", expectedOutput, false)
	}

	it should "reject invocation when user argument is missing" in {
	val expectedOutput = JsObject(
	"error" -> JsString("You must supply a 'user' parameter to authenticate with Message Hub.")
	)

	runActionWithExpectedResult(actionName, "dat/missingUser.json", expectedOutput, false)
	}

	it should "reject invocation when password argument is missing" in {
	val expectedOutput = JsObject(
	"error" -> JsString("You must supply a 'password' parameter to authenticate with Message Hub.")
	)

	runActionWithExpectedResult(actionName, "dat/missingPassword.json", expectedOutput, false)
	}

	it should "reject invocation when isJSONData and isBinaryValue are both enable" in {
	val expectedOutput = JsObject(
	"error" -> JsString("isJSONData and isBinaryValue cannot both be enabled."))

	runActionWithExpectedResult(actionName, "dat/multipleValueTypes.json", expectedOutput, false)
	}

	it should "fire a trigger when a binary message is posted to message hub" in withAssetCleaner(wskprops) {
	val currentTime = s"${System.currentTimeMillis}"

	(wp, assetHelper) =>
	val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
	println(s"Creating trigger ${triggerName}")

	createTrigger(assetHelper, triggerName, parameters = Map(
	"user" -> kafkaUtils.getAsJson("user"),
	"password" -> kafkaUtils.getAsJson("password"),
	"api_key" -> kafkaUtils.getAsJson("api_key"),
	"kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
	"kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
	"topic" -> topic.toJson,
	"isBinaryKey" -> true.toJson,
	"isBinaryValue" -> true.toJson))

	val defaultActionName = s"helloKafka-${currentTime}"

	assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
	action.create(name, defaultAction)
	}
	assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
	rule.create(name, trigger = triggerName, action = defaultActionName)
	}

	// It takes a moment for the consumer to fully initialize.
	println("Giving the consumer a moment to get ready")
	Thread.sleep(consumerInitTime)

	// key to use for the produced message
	val key = "TheKey"
	val encodedCurrentTime = Base64.getEncoder.encodeToString(currentTime.getBytes(StandardCharsets.UTF_8))
	val encodedKey = Base64.getEncoder.encodeToString(key.getBytes(StandardCharsets.UTF_8))

	withActivation(wsk.activation, wsk.action.invoke(s"$messagingPackage/$messageHubProduce", Map(
	"user" -> kafkaUtils.getAsJson("user"),
	"password" -> kafkaUtils.getAsJson("password"),
	"kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
	"topic" -> topic.toJson,
	"key" -> key.toJson,
	"value" -> currentTime.toJson))) {
	_.response.success shouldBe true
	}

	retry({
	println("Polling for activations")
	val activations = wsk.activation.pollFor(N = 1, Some(triggerName), retries = maxRetries)
	assert(activations.length == 1)

	val matchingActivations = for {
	id <- activations
	activation = wsk.activation.waitForActivation(id)
	if (activation.isRight && activation.right.get.fields.get("response").toString.contains(encodedCurrentTime))
	} yield activation.right.get

	assert(matchingActivations.length > 0)

	val activation = matchingActivations.head
	activation.getFieldPath("response", "success") shouldBe Some(true.toJson)

	// assert that there exists a message in the activation which has the expected keys and values
	val messages = KafkaUtils.messagesInActivation(activation, field = "value", value = encodedCurrentTime)
	assert(messages.length == 1)

	val message = messages.head
	message.getFieldPath("topic") shouldBe Some(topic.toJson)
	message.getFieldPath("key") shouldBe Some(encodedKey.toJson)
	}, N = 3)
	}

	it should "not fire a single trigger with an oversized payload" in withAssetCleaner(wskprops) {
	// payload size should be under the payload limit, but greater than 50% of the limit
	val testPayloadSize = 600000

	// create trigger
	val currentTime = s"${System.currentTimeMillis}"

	(wp, assetHelper) =>
	val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
	println(s"Creating trigger ${triggerName}")

	createTrigger(assetHelper, triggerName, parameters = Map(
	"user" -> kafkaUtils.getAsJson("user"),
	"password" -> kafkaUtils.getAsJson("password"),
	"api_key" -> kafkaUtils.getAsJson("api_key"),
	"kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
	"kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
	"topic" -> topic.toJson,
	"isBinaryKey" -> false.toJson,
	"isBinaryValue" -> false.toJson))

	val defaultActionName = s"helloKafka-${currentTime}"

	assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
	action.create(name, defaultAction)
	}
	assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
	rule.create(name, trigger = triggerName, action = defaultActionName)
	}

	// It takes a moment for the consumer to fully initialize.
	println("Giving the consumer a moment to get ready")
	Thread.sleep(consumerInitTime)

	// Rapidly produce two messages whose size are each greater than half the allowed payload limit.
	// This should ensure that the feed fires these as two separate triggers.
	println("Rapidly producing two large messages")
	val producer = kafkaUtils.createProducer()
	val firstMessage = new ProducerRecord(topic, "key", generateMessage(s"first${currentTime}", testPayloadSize))
	val secondMessage = new ProducerRecord(topic, "key", generateMessage(s"second${currentTime}", testPayloadSize))
	producer.send(firstMessage)
	producer.send(secondMessage)
	producer.close()

	retry({
	// verify there are two trigger activations required to handle these messages
	println("Polling for activations")
	val activations = wsk.activation.pollFor(N = 2, Some(triggerName), retries = maxRetries)

	println("Verifying activation content")
	val matchingActivations = for {
	id <- activations
	activation = wsk.activation.waitForActivation(id)
	if (activation.isRight && (activation.right.get.fields.get("response").toString.contains(s"first${currentTime}") \|\|
	activation.right.get.fields.get("response").toString.contains(s"second${currentTime}")))
	} yield activation.right.get

	assert(matchingActivations.length == 2)
	}, N = 3)
	}

	it should "not fire a trigger for a single oversized message" in withAssetCleaner(wskprops) {
	// payload size should be over the payload limit
	val testPayloadSize = 2000000

	// create trigger
	val currentTime = s"${System.currentTimeMillis}"

	(wp, assetHelper) =>
	val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
	println(s"Creating trigger ${triggerName}")

	createTrigger(assetHelper, triggerName, parameters = Map(
	"user" -> kafkaUtils.getAsJson("user"),
	"password" -> kafkaUtils.getAsJson("password"),
	"api_key" -> kafkaUtils.getAsJson("api_key"),
	"kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
	"kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
	"topic" -> topic.toJson,
	"isBinaryKey" -> false.toJson,
	"isBinaryValue" -> false.toJson))

	val defaultActionName = s"helloKafka-${currentTime}"

	assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
	action.create(name, defaultAction)
	}
	assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
	rule.create(name, trigger = triggerName, action = defaultActionName)
	}

	// It takes a moment for the consumer to fully initialize.
	println("Giving the consumer a moment to get ready")
	Thread.sleep(consumerInitTime)

	println("Producing an oversized message")
	val producer = kafkaUtils.createProducer()
	val bigMessage = new ProducerRecord(topic, "key", generateMessage(s"${currentTime}", testPayloadSize))
	producer.send(bigMessage)
	producer.close()

	retry({
	// verify there are no activations that match
	println("Polling for activations")
	val activations = wsk.activation.pollFor(N = 1, Some(triggerName), retries = maxRetries)

	println("Verifying activation content")
	val matchingActivations = for {
	id <- activations
	activation = wsk.activation.waitForActivation(id)
	if (activation.isRight && (activation.right.get.fields.get("response").toString.contains(s"first${currentTime}")))
	} yield activation.right.get

	assert(matchingActivations.length == 0)
	}, N = 3)
	}

	it should "reject trigger update without passing in any updatable parameters" in withAssetCleaner(wskprops) {
	val currentTime = s"${System.currentTimeMillis}"

	(wp, assetHelper) =>
	val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
	println(s"Creating trigger ${triggerName}")

	val username = kafkaUtils.getAsJson("user")
	val password = kafkaUtils.getAsJson("password")
	val admin_url = kafkaUtils.getAsJson("kafka_admin_url")
	val brokers = kafkaUtils.getAsJson("brokers")

	createTrigger(assetHelper, triggerName, parameters = Map(
	"user" -> username,
	"password" -> password,
	"api_key" -> kafkaUtils.getAsJson("api_key"),
	"kafka_admin_url" -> admin_url,
	"kafka_brokers_sasl" -> brokers,
	"topic" -> topic.toJson,
	"isBinaryKey" -> false.toJson,
	"isBinaryValue" -> false.toJson
	))

	val run = wsk.action.invoke(actionName, parameters = Map(
	"triggerName" -> triggerName.toJson,
	"lifecycleEvent" -> "UPDATE".toJson,
	"authKey" -> wp.authKey.toJson
	))

	withActivation(wsk.activation, run) {
	_.response.success shouldBe false
	}
	}

	it should "reject trigger update when both isJSONData and isBinaryValue are enabled" in withAssetCleaner(wskprops) {
	val currentTime = s"${System.currentTimeMillis}"

	(wp, assetHelper) =>
	val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
	println(s"Creating trigger $triggerName")

	val username = kafkaUtils.getAsJson("user")
	val password = kafkaUtils.getAsJson("password")
	val admin_url = kafkaUtils.getAsJson("kafka_admin_url")
	val brokers = kafkaUtils.getAsJson("brokers")

	createTrigger(assetHelper, triggerName, parameters = Map(
	"user" -> username,
	"password" -> password,
	"api_key" -> kafkaUtils.getAsJson("api_key"),
	"kafka_admin_url" -> admin_url,
	"kafka_brokers_sasl" -> brokers,
	"topic" -> topic.toJson,
	"isJSONData" -> true.toJson,
	"isBinaryKey" -> false.toJson,
	"isBinaryValue" -> false.toJson
	))

	val run = wsk.action.invoke(actionName, parameters = Map(
	"triggerName" -> triggerName.toJson,
	"lifecycleEvent" -> "UPDATE".toJson,
	"authKey" -> wp.authKey.toJson,
	"isBinaryValue" -> true.toJson
	))

	withActivation(wsk.activation, run) {
	_.response.success shouldBe false
	}
	}

	it should "fire a trigger when a message is posted to message hub before and after update" in withAssetCleaner(wskprops) {
	val currentTime = s"${System.currentTimeMillis}"

	(wp, assetHelper) =>
	val key = "TheKey"
	val triggerName = s"/_/dummyMessageHubTrigger-$currentTime"
	println(s"Creating trigger $triggerName")

	createTrigger(assetHelper, triggerName, parameters = Map(
	"user" -> kafkaUtils.getAsJson("user"),
	"password" -> kafkaUtils.getAsJson("password"),
	"api_key" -> kafkaUtils.getAsJson("api_key"),
	"kafka_admin_url" -> kafkaUtils.getAsJson("kafka_admin_url"),
	"kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
	"topic" -> topic.toJson
	))

	val defaultActionName = s"helloKafka-${currentTime}"

	assetHelper.withCleaner(wsk.action, defaultActionName) { (action, name) =>
	action.create(name, defaultAction)
	}
	assetHelper.withCleaner(wsk.rule, "rule") { (rule, name) =>
	rule.create(name, trigger = triggerName, action = defaultActionName)
	}

	println("Giving the consumer a moment to get ready")
	Thread.sleep(consumerInitTime)

	val first = Instant.now(Clock.systemUTC())

	println("Producing a message")
	withActivation(wsk.activation, wsk.action.invoke(s"$messagingPackage/$messageHubProduce", Map(
	"user" -> kafkaUtils.getAsJson("user"),
	"password" -> kafkaUtils.getAsJson("password"),
	"kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
	"topic" -> topic.toJson,
	"key" -> key.toJson,
	"value" -> currentTime.toJson
	))) {
	_.response.success shouldBe true
	}

	checkForActivations(triggerName, first, topic, key, currentTime)

	println("Updating trigger")

	val updateRunResult = wsk.action.invoke(actionName, parameters = Map(
	"triggerName" -> triggerName.toJson,
	"lifecycleEvent" -> "UPDATE".toJson,
	"authKey" -> wp.authKey.toJson,
	"isBinaryValue" -> true.toJson
	))

	withActivation(wsk.activation, updateRunResult) {
	_.response.success shouldBe true
	}

	println("Giving the consumer a moment to get ready")
	Thread.sleep(consumerInitTime)

	val second = Instant.now(Clock.systemUTC())
	val encodedCurrentTime = Base64.getEncoder.encodeToString(currentTime.getBytes(StandardCharsets.UTF_8))

	println("Producing a message")
	withActivation(wsk.activation, wsk.action.invoke(s"$messagingPackage/$messageHubProduce", Map(
	"user" -> kafkaUtils.getAsJson("user"),
	"password" -> kafkaUtils.getAsJson("password"),
	"kafka_brokers_sasl" -> kafkaUtils.getAsJson("brokers"),
	"topic" -> topic.toJson,
	"key" -> key.toJson,
	"value" -> currentTime.toJson
	))) {
	_.response.success shouldBe true
	}

	checkForActivations(triggerName, second, topic, key, encodedCurrentTime)
	}

	def createTrigger(assetHelper: AssetCleaner, name: String, parameters: Map[String, spray.json.JsValue]) = {
	val feedCreationResult = assetHelper.withCleaner(wsk.trigger, name) {
	(trigger, _) =>
	trigger.create(name, feed = Some(s"$messagingPackage/$messageHubFeed"), parameters = parameters)
	}

	withActivation(wsk.activation, feedCreationResult, initialWait = 5 seconds, totalWait = 60 seconds) {
	activation =>
	// should be successful
	activation.response.success shouldBe true
	}
	}

	def checkForActivations(triggerName: String, since: Instant, topic: String, key: String, value: String) = {
	retry({
	println("Polling for activations")
	val activations = wsk.activation.pollFor(N = 1, Some(triggerName), since = Some(since), retries = maxRetries)
	assert(activations.length == 1)

	println("Validating content of activation(s)")
	val matchingActivations = for {
	id <- activations
	activation = wsk.activation.waitForActivation(id)
	if (activation.isRight && activation.right.get.fields.get("response").toString.contains(value))
	} yield activation.right.get

	assert(matchingActivations.length > 0)

	val activation = matchingActivations.head
	activation.getFieldPath("response", "success") shouldBe Some(true.toJson)

	// assert that there exists a message in the activation which has the expected keys and values
	val messages = KafkaUtils.messagesInActivation(activation, field = "value", value = value)
	assert(messages.length == 1)

	val message = messages.head
	message.getFieldPath("topic") shouldBe Some(topic.toJson)
	message.getFieldPath("key") shouldBe Some(key.toJson)
	}, N = 3)
	}

	def generateMessage(prefix: String, size: Int): String = {
	val longString = Array.fill[String](size)("0").mkString("")
	s"${prefix}${longString}"
	}
	}