Apply comments

common/scala/src/main/scala/org/apache/openwhisk/common/AverageRingBuffer.scala

@@ -1,3 +1,20 @@
+/*
+ * 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 org.apache.openwhisk.common
 
 object AverageRingBuffer {

core/scheduler/src/main/resources/application.conf

@@ -0,0 +1,83 @@
+#
+# 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.
+#
+
+
+akka {
+  extensions = ["com.romix.akka.serialization.kryo.KryoSerializationExtension$"]
+  actor {
+    allow-java-serialization = off
+    serializers {
+      kryo = "com.romix.akka.serialization.kryo.KryoSerializer"
+    }
+    serialization-bindings {
+      "org.apache.openwhisk.core.scheduler.queue.CreateQueue" = kryo
+      "org.apache.openwhisk.core.scheduler.queue.CreateQueueResponse" = kryo
+      "org.apache.openwhisk.core.connector.ActivationMessage" = kryo
+    }
+    kryo {
+      idstrategy = "automatic"
+      classes = [
+        "org.apache.openwhisk.core.scheduler.queue.CreateQueue",
+        "org.apache.openwhisk.core.scheduler.queue.CreateQueueResponse",
+        "org.apache.openwhisk.core.connector.ActivationMessage"
+      ]
+    }
+  }
+
+  remote.netty.tcp {
+    send-buffer-size = 3151796b
+    receive-buffer-size = 3151796b
+    maximum-frame-size = 3151796b
+  }
+}
+
+whisk {
+  # tracing configuration
+  tracing {
+    component = "Scheduler"
+  }
+
+  fraction {
+      managed-fraction: 90%
+      blackbox-fraction: 10%
+  }
+
+  scheduler {
+    protocol = "http"
+    username: "scheduler.user"
+    password: "scheduler.pass"
+    grpc {
+      tls = "false"
+    }
+    queue {
+      idle-grace = "20 seconds"
+      stop-grace = "20 seconds"
+      flush-grace = "60 seconds"
+      graceful-shutdown-timeout = "5 seconds"
+      max-retention-size = "10000"
+      max-retention-ms = "60000"
+      throttling-fraction = "0.9"
+      duration-buffer-size = "10"
+    }
+    queue-manager {
+      max-scheduling-time-ms = "20000"
+      max-retries-to-get-queue = "13"
+    }
+    max-peek = "128"
+    in-progress-job-retention = "20"
+  }
+}

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/MemoryQueue.scala

@@ -626,6 +626,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
   }
 
   private def cleanUpActorsAndGotoRemovedIfPossible(data: RemovingData) = {
+    requestBuffer = requestBuffer.filter(!_.promise.isCompleted)
     if (queue.isEmpty && requestBuffer.isEmpty) {
       logging.info(this, s"[$invocationNamespace:$action:$stateName] No activation exist. Shutdown the queue.")
       // it can be safely called multiple times as it's idempotent
@@ -642,7 +643,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
     } else {
       logging.info(
         this,
-        s"[$invocationNamespace:$action:$stateName] Queue is going to stop but there are still ${queue.size} activations and ${requestBuffer.size} requst buffered.")
+        s"[$invocationNamespace:$action:$stateName] Queue is going to stop but there are still ${queue.size} activations and ${requestBuffer.size} request buffered.")
       stay // waiting for next timeout
     }
   }