Add task retries smoke product test

testing/trino-product-tests-groups/src/main/java/io/trino/tests/product/SuiteGroups.java

@@ -19,6 +19,7 @@
 
 import static io.trino.tests.product.TestGroups.AZURE;
 import static io.trino.tests.product.TestGroups.CLI;
+import static io.trino.tests.product.TestGroups.FAULT_TOLERANT;
 import static io.trino.tests.product.TestGroups.FUNCTIONS;
 import static io.trino.tests.product.TestGroups.HIVE_COMPRESSION;
 import static io.trino.tests.product.TestGroups.JDBC;
@@ -41,6 +42,7 @@ private SuiteGroups() {}
             JDBC,
             TRINO_JDBC,
             JDBC_KERBEROS_CONSTRAINED_DELEGATION,
+            FAULT_TOLERANT,
             FUNCTIONS,
             HIVE_COMPRESSION,
             LARGE_QUERY,

testing/trino-product-tests-groups/src/main/java/io/trino/tests/product/TestGroups.java

@@ -104,6 +104,7 @@ public final class TestGroups
     public static final String HUDI = "hudi";
     public static final String PARQUET = "parquet";
     public static final String IGNITE = "ignite";
+    public static final String FAULT_TOLERANT = "fault-tolerant";
 
     private TestGroups() {}
 

testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/EnvironmentContainers.java

@@ -44,6 +44,11 @@ public static boolean isTrinoContainer(String name)
         return name.startsWith(TRINO);
     }
 
+    public static boolean isTrinoWorker(String name)
+    {
+        return name.startsWith(WORKER);
+    }
+
     /**
      * Use this method only when you place `tempto-configuration.yaml' in environment configuration directory.
      */

testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/EnvironmentModule.java

@@ -30,6 +30,7 @@
 import io.trino.tests.product.launcher.env.common.Minio;
 import io.trino.tests.product.launcher.env.common.Standard;
 import io.trino.tests.product.launcher.env.common.StandardMultinode;
+import io.trino.tests.product.launcher.env.common.TaskRetriesMultinode;
 import io.trino.tests.product.launcher.env.jdk.BuiltInJdkProvider;
 import io.trino.tests.product.launcher.env.jdk.DistributionDownloadingJdkProvider;
 import io.trino.tests.product.launcher.env.jdk.JdkProvider;
@@ -82,6 +83,7 @@ public void configure(Binder binder)
         binder.bind(KafkaSaslPlaintext.class).in(SINGLETON);
         binder.bind(Standard.class).in(SINGLETON);
         binder.bind(StandardMultinode.class).in(SINGLETON);
+        binder.bind(TaskRetriesMultinode.class).in(SINGLETON);
         binder.bind(Kerberos.class).in(SINGLETON);
         binder.bind(Minio.class).in(SINGLETON);
 

testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/common/TaskRetriesMultinode.java

@@ -0,0 +1,61 @@
+/*
+ * 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.tests.product.launcher.env.common;
+
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
+import io.trino.tests.product.launcher.docker.DockerFiles;
+import io.trino.tests.product.launcher.env.Environment;
+
+import java.util.List;
+
+import static io.trino.tests.product.launcher.env.EnvironmentContainers.COORDINATOR;
+import static io.trino.tests.product.launcher.env.EnvironmentContainers.isTrinoWorker;
+import static io.trino.tests.product.launcher.env.common.Standard.CONTAINER_TRINO_CONFIG_PROPERTIES;
+import static java.util.Objects.requireNonNull;
+import static org.testcontainers.utility.MountableFile.forHostPath;
+
+public class TaskRetriesMultinode
+        implements EnvironmentExtender
+{
+    private final StandardMultinode standardMultinode;
+    private final DockerFiles.ResourceProvider configDir;
+
+    @Inject
+    public TaskRetriesMultinode(
+            StandardMultinode standardMultinode,
+            DockerFiles dockerFiles)
+    {
+        this.standardMultinode = requireNonNull(standardMultinode, "standardMultinode is null");
+        this.configDir = dockerFiles.getDockerFilesHostDirectory("common/task-retries-multinode");
+    }
+
+    @Override
+    public List<EnvironmentExtender> getDependencies()
+    {
+        return ImmutableList.of(standardMultinode);
+    }
+
+    @Override
+    public void extendEnvironment(Environment.Builder builder)
+    {
+        builder.configureContainer(COORDINATOR, container -> container
+                .withCopyFileToContainer(forHostPath(configDir.getPath("multinode-master-config.properties")), CONTAINER_TRINO_CONFIG_PROPERTIES));
+        builder.configureContainers(container -> {
+            if (isTrinoWorker(container.getLogicalName())) {
+                container.withCopyFileToContainer(forHostPath(configDir.getPath("multinode-worker-config.properties")), CONTAINER_TRINO_CONFIG_PROPERTIES);
+            }
+        });
+    }
+}

testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/env/environment/EnvMultinodeMinioDataLakeTaskRetriesFilesystem.java

@@ -0,0 +1,88 @@
+/*
+ * 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.tests.product.launcher.env.environment;
+
+import com.google.inject.Inject;
+import io.trino.tests.product.launcher.docker.DockerFiles;
+import io.trino.tests.product.launcher.env.Environment;
+import io.trino.tests.product.launcher.env.EnvironmentProvider;
+import io.trino.tests.product.launcher.env.common.Hadoop;
+import io.trino.tests.product.launcher.env.common.Minio;
+import io.trino.tests.product.launcher.env.common.TaskRetriesMultinode;
+import io.trino.tests.product.launcher.env.common.TestsEnvironment;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileAttribute;
+import java.nio.file.attribute.PosixFilePermission;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.util.Set;
+
+import static io.trino.tests.product.launcher.env.EnvironmentContainers.TESTS;
+import static io.trino.tests.product.launcher.env.EnvironmentContainers.isTrinoContainer;
+import static io.trino.tests.product.launcher.env.common.Minio.MINIO_CONTAINER_NAME;
+import static io.trino.tests.product.launcher.env.common.Standard.CONTAINER_TRINO_ETC;
+import static org.testcontainers.utility.MountableFile.forHostPath;
+
+/**
+ * Trino with S3-compatible Data Lake setup based on MinIO.
+ * Task retries enabled using Filesystem exchange backed by MinIO.
+ */
+@TestsEnvironment
+public class EnvMultinodeMinioDataLakeTaskRetriesFilesystem
+        extends EnvironmentProvider
+{
+    private static final String S3_BUCKET_NAME = "test-bucket";
+
+    private final DockerFiles.ResourceProvider configDir;
+
+    @Inject
+    public EnvMultinodeMinioDataLakeTaskRetriesFilesystem(TaskRetriesMultinode taskRetriesMultinode, Hadoop hadoop, Minio minio, DockerFiles dockerFiles)
+    {
+        super(taskRetriesMultinode, hadoop, minio);
+        this.configDir = dockerFiles.getDockerFilesHostDirectory("conf/environment/multinode-minio-data-lake-task-retries-filesystem");
+    }
+
+    @Override
+    public void extendEnvironment(Environment.Builder builder)
+    {
+        builder.configureContainer(TESTS, dockerContainer -> {
+            dockerContainer.withEnv("S3_BUCKET", S3_BUCKET_NAME);
+        });
+
+        // initialize buckets in minio
+        FileAttribute<Set<PosixFilePermission>> posixFilePermissions = PosixFilePermissions.asFileAttribute(PosixFilePermissions.fromString("rw-r--r--"));
+        Path minioBucketDirectory;
+        try {
+            minioBucketDirectory = Files.createTempDirectory("test-bucket-contents", posixFilePermissions);
+            minioBucketDirectory.toFile().deleteOnExit();
+        }
+        catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+        builder.configureContainer(MINIO_CONTAINER_NAME, container ->
+                container.withCopyFileToContainer(forHostPath(minioBucketDirectory), "/data/" + S3_BUCKET_NAME));
+
+        builder.addConnector("iceberg", forHostPath(configDir.getPath("iceberg.properties")));
+
+        // configure exchange mananger
+        builder.configureContainers(container -> {
+            if (isTrinoContainer(container.getLogicalName())) {
+                container.withCopyFileToContainer(forHostPath(configDir.getPath("exchange-manager.properties")), CONTAINER_TRINO_ETC + "/exchange-manager.properties");
+            }
+        });
+    }
+}

testing/trino-product-tests-launcher/src/main/java/io/trino/tests/product/launcher/suite/suites/Suite7NonGeneric.java

@@ -17,6 +17,7 @@
 import io.trino.tests.product.launcher.env.EnvironmentConfig;
 import io.trino.tests.product.launcher.env.EnvironmentDefaults;
 import io.trino.tests.product.launcher.env.environment.EnvMultinodeKerberosKudu;
+import io.trino.tests.product.launcher.env.environment.EnvMultinodeMinioDataLakeTaskRetriesFilesystem;
 import io.trino.tests.product.launcher.env.environment.EnvMultinodePostgresql;
 import io.trino.tests.product.launcher.env.environment.EnvMultinodeSqlserver;
 import io.trino.tests.product.launcher.env.environment.EnvSinglenodeKerberosHdfsImpersonationCrossRealm;
@@ -32,6 +33,7 @@
 import static com.google.common.base.Verify.verify;
 import static io.trino.tests.product.TestGroups.CLI;
 import static io.trino.tests.product.TestGroups.CONFIGURED_FEATURES;
+import static io.trino.tests.product.TestGroups.FAULT_TOLERANT;
 import static io.trino.tests.product.TestGroups.HDFS_IMPERSONATION;
 import static io.trino.tests.product.TestGroups.HIVE_KERBEROS;
 import static io.trino.tests.product.TestGroups.HIVE_SPARK;
@@ -75,6 +77,9 @@ public List<SuiteTestRun> getTestRuns(EnvironmentConfig config)
                         .build(),
                 testOnEnvironment(EnvTwoKerberosHives.class)
                         .withGroups(CONFIGURED_FEATURES, TWO_HIVES)
+                        .build(),
+                testOnEnvironment(EnvMultinodeMinioDataLakeTaskRetriesFilesystem.class)
+                        .withGroups(FAULT_TOLERANT)
                         .build());
     }
 }

testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/task-retries-multinode/multinode-master-config.properties

@@ -0,0 +1,17 @@
+node.id=will-be-overwritten
+node.environment=test
+
+coordinator=true
+experimental.concurrent-startup=true
+node-scheduler.include-coordinator=false
+http-server.http.port=8080
+query.max-memory=1GB
+discovery.uri=http://presto-master:8080
+
+# Use task.min-writer-count > 1, as this allows to expose writer-concurrency related bugs.
+task.min-writer-count=2
+task.concurrency=2
+task.max-writer-count=2
+
+# Enable task retries
+retry-policy=task

testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/common/task-retries-multinode/multinode-worker-config.properties

@@ -0,0 +1,17 @@
+node.id=will-be-overwritten
+node.environment=test
+
+coordinator=false
+experimental.concurrent-startup=true
+http-server.http.port=8081
+query.max-memory=1GB
+query.max-memory-per-node=1GB
+discovery.uri=http://presto-master:8080
+
+# Use task.min-writer-count > 1, as this allows to expose writer-concurrency related bugs.
+task.min-writer-count=2
+task.concurrency=2
+task.max-writer-count=2
+
+# Enable task retries
+retry-policy=task

testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-minio-data-lake-task-retries-filesystem/exchange-manager.properties

@@ -0,0 +1,7 @@
+exchange-manager.name=filesystem
+exchange.s3.endpoint=http://minio:9080/
+exchange.s3.region=us-west-1
+exchange.s3.aws-access-key=minio-access-key
+exchange.s3.aws-secret-key=minio-secret-key
+exchange.s3.path-style-access=true
+exchange.base-directories=s3://test-bucket

testing/trino-product-tests-launcher/src/main/resources/docker/presto-product-tests/conf/environment/multinode-minio-data-lake-task-retries-filesystem/iceberg.properties

@@ -0,0 +1,10 @@
+connector.name=iceberg
+hive.metastore.uri=thrift://hadoop-master:9083
+fs.native-s3.enabled=true
+fs.hadoop.enabled=false
+s3.region=us-east-1
+s3.aws-access-key=minio-access-key
+s3.aws-secret-key=minio-secret-key
+s3.endpoint=http://minio:9080/
+s3.path-style-access=true
+iceberg.file-format=PARQUET

testing/trino-product-tests/src/main/java/io/trino/tests/product/faulttolerant/TestTaskRetriesFilesystemSmoke.java

@@ -0,0 +1,32 @@
+/*
+ * 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.tests.product.faulttolerant;
+
+import io.trino.tempto.ProductTest;
+import org.testng.annotations.Test;
+
+import static io.trino.tests.product.TestGroups.FAULT_TOLERANT;
+import static io.trino.tests.product.utils.QueryExecutors.onTrino;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class TestTaskRetriesFilesystemSmoke
+        extends ProductTest
+{
+    @Test(groups = FAULT_TOLERANT)
+    public void testSimpleQuery()
+    {
+        assertThat(onTrino().executeQuery("select count(*) from tpch.sf1.lineitem where orderkey < 100").getOnlyValue())
+                .isEqualTo(105L);
+    }
+}