Enable HTTP/2 for internal communication by default

client/trino-client/src/main/java/io/trino/client/OkHttpUtil.java

@@ -15,7 +15,6 @@
 
 import com.google.common.base.CharMatcher;
 import com.google.common.base.StandardSystemProperty;
-import com.google.common.collect.ImmutableList;
 import com.google.common.net.HostAndPort;
 import io.trino.client.auth.kerberos.DelegatedConstrainedContextProvider;
 import io.trino.client.auth.kerberos.DelegatedUnconstrainedContextProvider;
@@ -27,7 +26,6 @@
 import okhttp3.Interceptor;
 import okhttp3.JavaNetCookieJar;
 import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
 import okhttp3.internal.tls.LegacyHostnameVerifier;
 import okhttp3.logging.HttpLoggingInterceptor;
 import okhttp3.logging.HttpLoggingInterceptor.Level;
@@ -123,12 +121,6 @@ public static void setupCookieJar(OkHttpClient.Builder clientBuilder)
         clientBuilder.cookieJar(new JavaNetCookieJar(new CookieManager()));
     }
 
-    public static void disableHttp2(OkHttpClient.Builder clientBuilder)
-    {
-        // Disable HTTP/2 as it's not tested nor supported
-        clientBuilder.protocols(ImmutableList.of(Protocol.HTTP_1_1, Protocol.HTTP_1_1));
-    }
-
     public static void setupSocksProxy(OkHttpClient.Builder clientBuilder, Optional<HostAndPort> socksProxy)
     {
         setupProxy(clientBuilder, socksProxy, SOCKS);

client/trino-client/src/main/java/io/trino/client/uri/HttpClientFactory.java

@@ -29,7 +29,6 @@
 
 import static io.trino.client.KerberosUtil.defaultCredentialCachePath;
 import static io.trino.client.OkHttpUtil.basicAuth;
-import static io.trino.client.OkHttpUtil.disableHttp2;
 import static io.trino.client.OkHttpUtil.setupAlternateHostnameVerification;
 import static io.trino.client.OkHttpUtil.setupCookieJar;
 import static io.trino.client.OkHttpUtil.setupHttpLogging;
@@ -53,7 +52,6 @@ private HttpClientFactory() {}
     public static OkHttpClient.Builder toHttpClientBuilder(TrinoUri uri, String userAgent)
     {
         OkHttpClient.Builder builder = unauthenticatedClientBuilder(uri, userAgent);
-        disableHttp2(builder);
         setupCookieJar(builder);
 
         if (!uri.isUseSecureConnection()) {

core/trino-main/src/main/java/io/trino/TrinoMediaTypes.java

@@ -17,8 +17,8 @@
 
 public final class TrinoMediaTypes
 {
-    public static final String TRINO_PAGES = "application/X-trino-pages";
-    public static final MediaType TRINO_PAGES_TYPE = MediaType.create("application", "X-trino-pages");
+    public static final String TRINO_PAGES = "application/x-trino-pages";
+    public static final MediaType TRINO_PAGES_TYPE = MediaType.create("application", "x-trino-pages");
 
     private TrinoMediaTypes()
     {

core/trino-main/src/main/java/io/trino/dispatcher/QueuedStatementResource.java

@@ -32,7 +32,6 @@
 import io.trino.execution.ExecutionFailureInfo;
 import io.trino.execution.QueryManagerConfig;
 import io.trino.execution.QueryState;
-import io.trino.server.DisconnectionAwareAsyncResponse;
 import io.trino.server.ExternalUriInfo;
 import io.trino.server.GoneException;
 import io.trino.server.HttpRequestSessionContextFactory;
@@ -61,6 +60,7 @@
 import jakarta.ws.rs.PathParam;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.container.AsyncResponse;
 import jakarta.ws.rs.container.Suspended;
 import jakarta.ws.rs.core.Context;
 import jakarta.ws.rs.core.HttpHeaders;
@@ -84,10 +84,10 @@
 import static com.google.common.util.concurrent.Futures.nonCancellationPropagating;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static io.airlift.concurrent.Threads.daemonThreadsNamed;
+import static io.airlift.jaxrs.AsyncResponseHandler.bindAsyncResponse;
 import static io.trino.client.ProtocolHeaders.TRINO_HEADERS;
 import static io.trino.execution.QueryState.FAILED;
 import static io.trino.execution.QueryState.QUEUED;
-import static io.trino.server.DisconnectionAwareAsyncResponse.bindDisconnectionAwareAsyncResponse;
 import static io.trino.server.ServletSecurityUtils.authenticatedIdentity;
 import static io.trino.server.ServletSecurityUtils.clearAuthenticatedIdentity;
 import static io.trino.server.protocol.QueryInfoUrlFactory.getQueryInfoUri;
@@ -201,12 +201,12 @@ public void getStatus(
             @PathParam("token") long token,
             @QueryParam("maxWait") Duration maxWait,
             @BeanParam ExternalUriInfo externalUriInfo,
-            @Suspended @BeanParam DisconnectionAwareAsyncResponse asyncResponse)
+            @Suspended AsyncResponse asyncResponse)
     {
         Query query = getQuery(queryId, slug, token);
 
         ListenableFuture<Response> future = getStatus(query, token, maxWait, externalUriInfo);
-        bindDisconnectionAwareAsyncResponse(asyncResponse, future, responseExecutor);
+        bindAsyncResponse(asyncResponse, future, responseExecutor);
     }
 
     private ListenableFuture<Response> getStatus(Query query, long token, Duration maxWait, ExternalUriInfo externalUriInfo)

core/trino-main/src/main/java/io/trino/execution/TaskManagerConfig.java

@@ -74,7 +74,7 @@ public class TaskManagerConfig
     private Duration clientTimeout = new Duration(2, TimeUnit.MINUTES);
     private Duration infoMaxAge = new Duration(15, TimeUnit.MINUTES);
 
-    private Duration statusRefreshMaxWait = new Duration(1, TimeUnit.SECONDS);
+    private Duration statusRefreshMaxWait = new Duration(3, TimeUnit.SECONDS);
     private Duration infoUpdateInterval = new Duration(3, TimeUnit.SECONDS);
     private Duration taskTerminationTimeout = new Duration(1, TimeUnit.MINUTES);
 
@@ -122,7 +122,7 @@ public boolean isThreadPerDriverSchedulerEnabled()
     }
 
     @MinDuration("1ms")
-    @MaxDuration("10s")
+    @MaxDuration("60s")
     @NotNull
     public Duration getStatusRefreshMaxWait()
     {

core/trino-main/src/main/java/io/trino/operator/HttpPageBufferClient.java

@@ -80,7 +80,7 @@
 import static io.trino.server.InternalHeaders.TRINO_PAGE_TOKEN;
 import static io.trino.server.InternalHeaders.TRINO_TASK_FAILED;
 import static io.trino.server.InternalHeaders.TRINO_TASK_INSTANCE_ID;
-import static io.trino.server.PagesResponseWriter.SERIALIZED_PAGES_MAGIC;
+import static io.trino.server.PagesInputStreamFactory.SERIALIZED_PAGES_MAGIC;
 import static io.trino.spi.HostAddress.fromUri;
 import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static io.trino.spi.StandardErrorCode.REMOTE_BUFFER_CLOSE_FAILED;
@@ -528,7 +528,7 @@ public void onSuccess(@Nullable StatusResponse result)
             {
                 assertNotHoldsLock(HttpPageBufferClient.this);
 
-                if (result.getStatusCode() != NO_CONTENT.code()) {
+                if (result != null && result.getStatusCode() != NO_CONTENT.code()) {
                     onFailure(new TrinoTransportException(
                             REMOTE_BUFFER_CLOSE_FAILED,
                             fromUri(location),
@@ -607,11 +607,7 @@ public boolean equals(Object o)
 
         HttpPageBufferClient that = (HttpPageBufferClient) o;
 
-        if (!location.equals(that.location)) {
-            return false;
-        }
-
-        return true;
+        return location.equals(that.location);
     }
 
     @Override

core/trino-main/src/main/java/io/trino/server/AsyncResponseUtils.java

@@ -0,0 +1,36 @@
+/*
+ * 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 io.trino.server;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import io.airlift.units.Duration;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Supplier;
+
+import static com.google.common.util.concurrent.Futures.catching;
+import static com.google.common.util.concurrent.Futures.withTimeout;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+public class AsyncResponseUtils
+{
+    private AsyncResponseUtils() {}
+
+    public static <V> ListenableFuture<V> withFallbackAfterTimeout(ListenableFuture<V> future, Duration timeout, Supplier<V> fallback, Executor responseExecutor, ScheduledExecutorService timeoutExecutor)
+    {
+        return catching(withTimeout(future, timeout.toMillis(), MILLISECONDS, timeoutExecutor), TimeoutException.class, _ -> fallback.get(), responseExecutor);
+    }
+}

core/trino-main/src/main/java/io/trino/server/InternalCommunicationConfig.java

@@ -31,7 +31,7 @@
 public class InternalCommunicationConfig
 {
     private String sharedSecret;
-    private boolean http2Enabled;
+    private boolean http2Enabled = true;
     private boolean httpsRequired;
     private String keyStorePath;
     private String keyStorePassword;