fix: Correct exception handling for CloudEvents

A previous change resulted broke exception handling for cloud events so
that an exception from the function would not result in a failure
response code. Correct error handling for CloudEvents and add a test to
cover this case to avoid future breakages.

Author: maemayve

invoker/core/src/main/java/com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java

@@ -364,14 +364,7 @@ private <CloudEventT> void serviceCloudEvent(HttpServletRequest req) throws Exce
     // ServiceLoader.load
     // will throw ServiceConfigurationError. At this point we're still running with the default
     // context ClassLoader, which is the system ClassLoader that has loaded the code here.
-    try {
-      executionIdUtil.storeExecutionId(req);
-      runWithContextClassLoader(() -> executor.serviceCloudEvent(reader.toEvent(data -> data)));
-    } catch (Throwable t) {
-      logger.log(Level.SEVERE, "Failed to execute " + executor.functionName(), t);
-    } finally {
-      executionIdUtil.removeExecutionId();
-    }
+    runWithContextClassLoader(() -> executor.serviceCloudEvent(reader.toEvent(data -> data)));
     // The data->data is a workaround for a bug fixed since Milestone 4 of the SDK, in
     // https://github.com/cloudevents/sdk-java/pull/259.
   }

invoker/core/src/test/java/com/google/cloud/functions/invoker/IntegrationTest.java

@@ -577,6 +577,43 @@ public void nativeCloudEvent() throws Exception {
         ImmutableList.of(cloudEventsStructuredTestCase, cloudEventsBinaryTestCase));
   }
 
+  /** Tests a CloudEvent being handled by a CloudEvent handler throws exception */
+  @Test
+  public void nativeCloudEventException() throws Exception {
+    String exceptionExpectedOutput =
+        "\"severity\": \"ERROR\", \"logging.googleapis.com/sourceLocation\": {\"file\":"
+            + " \"com/google/cloud/functions/invoker/BackgroundFunctionExecutor.java\", \"method\":"
+            + " \"service\"}, \"execution_id\": \""
+            + EXECUTION_ID
+            + "\", "
+            + "\"message\": \"Failed to execute"
+            + " com.google.cloud.functions.invoker.testfunctions.ExceptionBackground\\n"
+            + "java.lang.RuntimeException: exception thrown for test";
+    File snoopFile = snoopFile();
+    CloudEvent cloudEvent = sampleCloudEvent(snoopFile);
+    EventFormat jsonFormat =
+        EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);
+    String cloudEventJson = new String(jsonFormat.serialize(cloudEvent), UTF_8);
+
+    // A CloudEvent using the "structured content mode", where both the metadata and the payload
+    // are in the body of the HTTP request.
+    TestCase cloudEventsStructuredTestCase =
+        TestCase.builder()
+            .setRequestText(cloudEventJson)
+            .setHttpContentType("application/cloudevents+json; charset=utf-8")
+            .setHttpHeaders(ImmutableMap.of(EXECUTION_ID_HTTP_HEADER, EXECUTION_ID))
+            .setExpectedResponseCode(500)
+            .setExpectedOutput(exceptionExpectedOutput)
+            .build();
+
+    testFunction(
+        SignatureType.CLOUD_EVENT,
+        fullTarget("ExceptionBackground"),
+        ImmutableList.of(),
+        ImmutableList.of(cloudEventsStructuredTestCase),
+        Collections.emptyMap());
+  }
+
   @Test
   public void nested() throws Exception {
     String testText = "sic transit gloria mundi";