chore(implementation)!: use Jetty-12 core without servlets

Port the invoker to upgrade to Eclipse Jetty-12 version 12.  Specifically using the new core APIs of Eclipse Jetty-12 that allow the overhead of a Servlet container to be avoided.

BREAKING CHANGE: use Java 17 or above, as required by Eclipse Jetty-12.

Signed-off-by: Lachlan Roberts <[email protected]>

Author: lachlan-roberts

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

@@ -185,9 +185,12 @@ static Optional<Type> backgroundFunctionTypeArgument(
   }
 
   private static Event parseLegacyEvent(Request req) throws IOException {
-    try (BufferedReader bodyReader = new BufferedReader(
-        new InputStreamReader(Content.Source.asInputStream(req),
-            Objects.requireNonNullElse(Request.getCharset(req), StandardCharsets.ISO_8859_1)))) {
+    try (BufferedReader bodyReader =
+        new BufferedReader(
+            new InputStreamReader(
+                Content.Source.asInputStream(req),
+                Objects.requireNonNullElse(
+                    Request.getCharset(req), StandardCharsets.ISO_8859_1)))) {
       return parseLegacyEvent(bodyReader);
     }
   }
@@ -367,6 +370,9 @@ private <CloudEventT> void serviceCloudEvent(Request req) throws Exception {
     @SuppressWarnings("unchecked")
     FunctionExecutor<CloudEventT> executor = (FunctionExecutor<CloudEventT>) functionExecutor;
 
+    // Read the entire request body into a byte array.
+    // TODO: this method is deprecated for removal, use the method introduced by
+    //    https://github.com/jetty/jetty.project/pull/13939 when it is released.
     byte[] body = Content.Source.asByteArrayAsync(req, -1).get();
     MessageReader reader = HttpMessageFactory.createReaderFromMultimap(headerMap(req), body);
     // It's important not to set the context ClassLoader earlier, because MessageUtils will use
@@ -384,7 +390,8 @@ private <CloudEventT> void serviceCloudEvent(Request req) throws Exception {
   private static Map<String, List<String>> headerMap(Request req) {
     Map<String, List<String>> headerMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
     for (HttpField field : req.getHeaders()) {
-      headerMap.computeIfAbsent(field.getName(), unused -> new ArrayList<>())
+      headerMap
+          .computeIfAbsent(field.getName(), unused -> new ArrayList<>())
           .addAll(field.getValueList());
     }
     return headerMap;

invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpRequestImpl.java

@@ -80,13 +80,14 @@ public Map<String, List<String>> getQueryParameters() {
   @Override
   public Map<String, HttpPart> getParts() {
     String contentType = request.getHeaders().get(HttpHeader.CONTENT_TYPE);
-    if (contentType == null || !contentType.startsWith(MimeTypes.Type.MULTIPART_FORM_DATA.asString())) {
+    if (contentType == null
+        || !contentType.startsWith(MimeTypes.Type.MULTIPART_FORM_DATA.asString())) {
       throw new IllegalStateException("Content-Type must be multipart/form-data: " + contentType);
     }
 
     // The multipart parsing is done by the EagerContentHandler, so we just call getParts.
     MultiPartFormData.Parts parts = MultiPartFormData.getParts(request);
-    if (parts == null){
+    if (parts == null) {
       throw new IllegalStateException();
     }
 
@@ -178,8 +179,8 @@ public Optional<String> getCharacterEncoding() {
 
     @Override
     public InputStream getInputStream() throws IOException {
-        Content.Source contentSource = part.createContentSource();
-        return Content.Source.asInputStream(contentSource);
+      Content.Source contentSource = part.createContentSource();
+      return Content.Source.asInputStream(contentSource);
     }
 
     @Override

invoker/core/src/main/java/com/google/cloud/functions/invoker/http/HttpResponseImpl.java

@@ -87,11 +87,15 @@ public OutputStream getOutputStream() {
     }
     if (outputStream == null) {
       Request request = response.getRequest();
-      int outputBufferSize = request.getConnectionMetaData().getHttpConfiguration()
-          .getOutputBufferSize();
-      BufferedContentSink bufferedContentSink = new BufferedContentSink(response,
-          request.getComponents().getByteBufferPool(),
-          false, outputBufferSize / 2, outputBufferSize);
+      int outputBufferSize =
+          request.getConnectionMetaData().getHttpConfiguration().getOutputBufferSize();
+      BufferedContentSink bufferedContentSink =
+          new BufferedContentSink(
+              response,
+              request.getComponents().getByteBufferPool(),
+              false,
+              outputBufferSize / 2,
+              outputBufferSize);
       outputStream = new ContentSinkOutputStream(bufferedContentSink);
     }
     return outputStream;
@@ -104,8 +108,10 @@ public synchronized BufferedWriter getWriter() throws IOException {
         throw new IllegalStateException("getOutputStream called");
       }
 
-      writer = new NonBufferedWriter(WriteThroughWriter.newWriter(getOutputStream(),
-          Objects.requireNonNullElse(charset, StandardCharsets.UTF_8)));
+      writer =
+          new NonBufferedWriter(
+              WriteThroughWriter.newWriter(
+                  getOutputStream(), Objects.requireNonNullElse(charset, StandardCharsets.UTF_8)));
     }
     return writer;
   }
@@ -130,13 +136,11 @@ public void close(Callback callback) {
   }
 
   /**
-   * A {@link BufferedWriter} that does not buffer.
-   * It is generally more efficient to buffer at the {@link Content.Sink} level,
-   * since frequently total content is smaller than a single buffer and
-   * the {@link Content.Sink} can turn a close into a last write that will avoid
-   * chunking the response if at all possible.   However, {@link BufferedWriter}
-   * is in the API for {@link HttpResponse}, so we must return a writer of
-   * that type.
+   * A {@link BufferedWriter} that does not buffer. It is generally more efficient to buffer at the
+   * {@link Content.Sink} level, since frequently total content is smaller than a single buffer and
+   * the {@link Content.Sink} can turn a close into a last write that will avoid chunking the
+   * response if at all possible. However, {@link BufferedWriter} is in the API for {@link
+   * HttpResponse}, so we must return a writer of that type.
    */
   private static class NonBufferedWriter extends BufferedWriter {
     private final Writer writer;

invoker/core/src/main/java/com/google/cloud/functions/invoker/http/TimeoutHandler.java

@@ -14,76 +14,65 @@
 
 package com.google.cloud.functions.invoker.http;
 
-import java.util.Timer;
-import java.util.TimerTask;
+import java.time.Duration;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
-import org.eclipse.jetty.http.HttpStatus;
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Response;
 import org.eclipse.jetty.util.Callback;
+import org.eclipse.jetty.util.thread.Scheduler;
 
 public class TimeoutHandler extends Handler.Wrapper {
-  private final int timeoutMs;
+  private final Duration timeout;
 
   public TimeoutHandler(int timeoutSeconds, Handler handler) {
     setHandler(handler);
-    this.timeoutMs = timeoutSeconds * 1000; // Convert seconds to milliseconds
+    timeout = Duration.ofSeconds(timeoutSeconds);
   }
 
   @Override
   public boolean handle(Request request, Response response, Callback callback) throws Exception {
-    // Wrap the callback to ensure it is only called once between the handler and the timeout task.
-    AtomicBoolean completed = new AtomicBoolean(false);
-    Callback wrappedCallback = new Callback() {
-      @Override
-      public void succeeded() {
-        if (completed.compareAndSet(false, true)) {
-          callback.succeeded();
-        }
-      }
+    // Wrap the callback to ensure it is only completed once between the
+    // handler and the timeout task.
+    Callback wrappedCallback = new ProtectedCallback(callback);
+    Scheduler.Task timeoutTask =
+        request
+            .getComponents()
+            .getScheduler()
+            .schedule(
+                () -> wrappedCallback.failed(new TimeoutException("Function execution timed out")),
+                timeout);
 
-      @Override
-      public void failed(Throwable x) {
-        if (completed.compareAndSet(false, true)) {
-            callback.failed(x);
-        }
-      }
+    // Cancel the timeout if the request completes the callback first.
+    return super.handle(request, response, Callback.from(timeoutTask::cancel, wrappedCallback));
+  }
 
-      @Override
-      public InvocationType getInvocationType() {
-        return callback.getInvocationType();
-      }
-    };
+  private static class ProtectedCallback implements Callback {
+    private final Callback callback;
+    private final AtomicBoolean completed = new AtomicBoolean(false);
 
-    // TODO: consider wrapping the request/response to throw if they are used after timeout.
-    // TODO: Use org.eclipse.jetty.io.CyclicTimeouts which is optimized for timeouts which are almost always cancelled.
-    Timer timer = new Timer(true);
-    TimerTask timeoutTask =
-        new TimerTask() {
-          @Override
-          public void run() {
-            // TODO: there is a race between the handler writing response and timeout firing.
-            //  This timeout firing doesn't stop the thread handling the request / response it just writes an error to the response.
-            Response.writeError(
-                request,
-                response,
-                callback,
-                HttpStatus.REQUEST_TIMEOUT_408,
-                "Function execution timed out");
-          }
-        };
+    public ProtectedCallback(Callback callback) {
+      this.callback = callback;
+    }
 
-    timer.schedule(timeoutTask, timeoutMs);
+    @Override
+    public void succeeded() {
+      if (completed.compareAndSet(false, true)) {
+        callback.succeeded();
+      }
+    }
 
-    boolean handle;
-    try {
-      handle = super.handle(request, response, wrappedCallback);
-      timeoutTask.cancel();
-    } finally {
-      timer.purge();
+    @Override
+    public void failed(Throwable x) {
+      if (completed.compareAndSet(false, true)) {
+        callback.failed(x);
+      }
     }
 
-    return handle;
+    @Override
+    public InvocationType getInvocationType() {
+      return callback.getInvocationType();
+    }
   }
-}
+}

invoker/core/src/main/java/com/google/cloud/functions/invoker/runner/Invoker.java

@@ -283,7 +283,13 @@ private void startServer(boolean join) throws Exception {
     server.setErrorHandler(
         new ErrorHandler() {
           @Override
-          protected void generateResponse(Request request, Response response, int code, String message, Throwable cause, Callback callback) {
+          protected void generateResponse(
+              Request request,
+              Response response,
+              int code,
+              String message,
+              Throwable cause,
+              Callback callback) {
             // Suppress error body
             callback.succeeded();
           }