feat(queue): Add QueueContextHolder (#4013)

Exposes the originating queue message to the worker thread executing on the message.
This intended for observability purposes.

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

keiko-core/src/main/kotlin/com/netflix/spinnaker/q/QueueContextHolder.kt

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2020 Netflix, Inc.
+ *
+ * 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 com.netflix.spinnaker.q
+
+/**
+ * Holds a [Message] object in a [ThreadLocal].
+ */
+object QueueContextHolder {
+
+  private val holder: ThreadLocal<Message> = ThreadLocal()
+
+  fun set(context: Message) {
+    holder.set(context)
+  }
+
+  fun get(): Message? = holder.get()
+
+  fun clear() {
+    holder.remove()
+  }
+}

keiko-core/src/main/kotlin/com/netflix/spinnaker/q/QueueProcessor.kt

@@ -86,12 +86,15 @@ class QueueProcessor(
       try {
         executor.execute {
           try {
+            QueueContextHolder.set(message)
             handler.invoke(message)
             ack.invoke()
           } catch (e: Throwable) {
             // Something very bad is happening
             log.error("Unhandled throwable from $message", e)
             publisher.publishEvent(HandlerThrewError(message))
+          } finally {
+            QueueContextHolder.clear()
           }
         }
       } catch (e: RejectedExecutionException) {