fix: correct Cloud Event retry functionality

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/main/java/com/google/cloud/functions/invoker/gcf/JsonLogHandler.java

@@ -17,6 +17,7 @@
  */
 public final class JsonLogHandler extends Handler {
   private static final String SOURCE_LOCATION_KEY = "\"logging.googleapis.com/sourceLocation\": ";
+  private static final String LOG_EXECUTION_ID_ENV_NAME = "LOG_EXECUTION_ID";
 
   private static final String DEBUG = "DEBUG";
   private static final String INFO = "INFO";
@@ -108,9 +109,11 @@ private static void appendSourceLocation(StringBuilder json, LogRecord record) {
   }
 
   private void appendExecutionId(StringBuilder json, LogRecord record) {
-    json.append("\"execution_id\": \"")
-        .append(executionIdByThreadMap.get(Integer.toString(record.getThreadID())))
-        .append("\", ");
+    if (executionIdLoggingEnabled()) {
+      json.append("\"execution_id\": \"")
+          .append(executionIdByThreadMap.get(Integer.toString(record.getThreadID())))
+          .append("\", ");
+    }
   }
 
   private static String escapeString(String s) {
@@ -142,4 +145,8 @@ public void addExecutionId(long threadId, String executionId) {
   public void removeExecutionId(long threadId) {
     executionIdByThreadMap.remove(Long.toString(threadId));
   }
+
+  private boolean executionIdLoggingEnabled() {
+    return Boolean.parseBoolean(System.getenv().getOrDefault(LOG_EXECUTION_ID_ENV_NAME, "false"));
+  }
 }

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";