Skip to content

Commit 37fd39b

Browse files
nevzhengnevzheng
committed
[#14002] improvement(test): Run Trino through fixtures and wait for it without commands in the container
1 parent 13f3f3b commit 37fd39b

323 files changed

Lines changed: 534 additions & 13781 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.github/workflows/trino-integration-test-action.yml‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -62,8 +62,6 @@ jobs:
6262
id: integrationTest
6363
run: |
6464
trino-connector/integration-test/trino-test-tools/trino_integration_test.sh
65-
# Disable the Trino cascading query integration test, because the connector jars are private now.
66-
#trino-connector/integration-test/trino-test-tools/run_test.sh
6765
6866
# The SQL-file harness above does not execute the JUnit ITs under
6967
# trino-connector/integration-test, so run them explicitly. This runs in embedded mode;
@@ -82,7 +80,6 @@ jobs:
8280
build/reports
8381
trino-connector/integration-test/build/*.log
8482
trino-connector/integration-test/build/*.tar
85-
trino-connector/integration-test/build/trino-cascading-env
8683
integration-test-common/build/trino-ci-container-log
8784
distribution/package/logs/gravitino-server.out
8885
distribution/package/logs/gravitino-server.log

‎docs/how-to-test.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,7 @@ To deploy the Gravitino server locally to run the integration tests, follow thes
110110
in the `distribution` directory.
111111
3. Use the `./gradlew test [--rerun-tasks] -PskipTests -PtestMode=deploy` command to run the
112112
integration tests in the `distribution` directory.
113-
4. Use the `bash trino-connector/integration-test/trino-test-tools/trino_test.sh` command to run all the
113+
4. Use the `bash trino-connector/integration-test/trino-test-tools/trino_integration_test.sh` command to run all the
114114
Trino test sets in the `trino-connector/integration-test/src/test/resources/trino-ci-testset/testsets` directory.
115115
Specify the `--trino_worker_num` parameter to make the Trino test sets run in a distributed environment.
116116
Specify the `--trino_version` parameter to make the Trino test sets run with specific trino version.

‎integration-test-common/docker-script/init/trino/init.sh‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,8 @@ if [[ "${TRINO_ROLE}" == "coordinator" ]]; then
4646
fi
4747
else
4848
cp "$trino_conf_dir/config/config-worker.properties" /etc/trino/config.properties
49+
# A private cluster's coordinator has a per-run host name, so it can share a network with others.
50+
sed -i "s|^discovery.uri=.*|discovery.uri=http://${TRINO_COORDINATOR_HOST:-trino}:8080|" /etc/trino/config.properties
4951
fi
5052

5153
# mkdir `node.data-dir` in the `/conf/node.properties`

‎integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/HiveContainer.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -134,7 +134,7 @@ public void start() {
134134
allowHadoopUserToImpersonate();
135135
}
136136
// Set here rather than in waitStrategy(), which runs before the constructor sets the auth.
137-
container.waitingFor(kerberosEnabled ? kerberosReadiness() : readiness());
137+
container.waitingFor(kerberosEnabled ? kerberosReadiness() : readiness(getHadoopUser()));
138138
withStartupTimeout(startupTimeout());
139139
super.start();
140140
}
@@ -216,9 +216,13 @@ private void outputHiveStatus() {
216216
* Waits for HDFS through the NameNode's JMX, creates the warehouse directory through WebHDFS and
217217
* waits for the metastore and HiveServer2 ports. The Hive CLI checks this replaces each boot a
218218
* JVM with an embedded metastore, which costs several seconds per call; the metastore port covers
219-
* Hive itself.
219+
* Hive itself. Any non-Kerberos container of the Hive image can wait with it, such as the Hive of
220+
* a {@link TrinoCluster}.
221+
*
222+
* @param hadoopUser the user that creates the warehouse directory, as the Hive CLI would have
223+
* @return the wait strategy
220224
*/
221-
private WaitStrategy readiness() {
225+
static WaitStrategy readiness(String hadoopUser) {
222226
AtomicInteger namenodePort = new AtomicInteger();
223227
return new WaitAllStrategy()
224228
.withStrategy(Wait.forListeningPort())
@@ -255,7 +259,7 @@ private WaitStrategy readiness() {
255259
ipAddress(target.getContainerInfo()),
256260
namenodePort.get(),
257261
WAREHOUSE_DIR,
258-
getHadoopUser()));
262+
hadoopUser));
259263
return response != null && response.contains("true");
260264
})
261265
.withPollInterval(Duration.ofMillis(500)))

‎integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/SharedContainers.java‎

Lines changed: 51 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
import java.util.concurrent.Future;
5151
import java.util.function.Consumer;
5252
import java.util.function.Function;
53+
import java.util.function.Supplier;
5354
import javax.annotation.Nullable;
5455
import org.apache.gravitino.integration.test.util.CloseableGroup;
5556
import org.apache.gravitino.integration.test.util.ITUtils;
@@ -91,6 +92,9 @@ class SharedContainers {
9192
private static volatile HiveContainer hiveRangerContainer;
9293
// One cluster per Gravitino server: a test class with another server replaces it.
9394
private static final ContainerSlot<TrinoSpec, TrinoCluster> trinoCluster = new ContainerSlot<>();
95+
// One standalone Trino node per Gravitino server and metalake.
96+
private static final ContainerSlot<TrinoNodeSpec, TrinoContainer> trinoNode =
97+
new ContainerSlot<>();
9498
private static volatile RangerContainer rangerContainer;
9599
private static volatile KafkaContainer kafkaContainer;
96100
private static final ContainerSlot<String, HiveContainer> kerberosHiveContainer =
@@ -775,20 +779,59 @@ TrinoCluster startTrinoCluster(TrinoSpec spec) {
775779
spec,
776780
() -> {
777781
initIfNecessary();
778-
TrinoCluster cluster =
779-
closer.register(
780-
TrinoCluster.builder()
781-
.withTrinoVersion(spec.version())
782-
.withWorkers(spec.workers())
783-
.withHiveRuntimeVersion(spec.hive().runtimeVersion())
784-
.withGravitinoServerPort(spec.gravitinoServerPort())
785-
.build());
782+
TrinoCluster cluster = closer.register(trinoClusterBuilder(spec).build());
786783
start(cluster);
787784
return cluster;
788785
});
789786
}
790787
}
791788

789+
/**
790+
* Starts a Trino cluster for the spec that only the caller uses. Its coordinator has a per-run
791+
* host name, so it runs beside the shared cluster; the caller closes it.
792+
*
793+
* @param spec the cluster to run, with the Gravitino server port set
794+
* @return the running private cluster
795+
*/
796+
TrinoCluster startPrivateTrinoCluster(TrinoSpec spec) {
797+
initIfNecessary();
798+
// The builder joins the run's network itself, or makes one of its own when there is none.
799+
return startPrivate(
800+
network,
801+
unused -> trinoClusterBuilder(spec).withPrivateNames(true).build(),
802+
SharedContainers::start,
803+
cluster -> {});
804+
}
805+
806+
static TrinoCluster.Builder trinoClusterBuilder(TrinoSpec spec) {
807+
return TrinoCluster.builder()
808+
.withTrinoVersion(spec.version())
809+
.withWorkers(spec.workers())
810+
.withHiveRuntimeVersion(spec.hive().runtimeVersion())
811+
.withGravitinoServerPort(spec.gravitinoServerPort());
812+
}
813+
814+
/**
815+
* Starts, or returns the running, standalone Trino node for the spec. A spec for another server
816+
* or metalake replaces the running node.
817+
*
818+
* @param spec the node to run
819+
* @param builder builds the node once its dependencies run
820+
* @return the running node
821+
*/
822+
TrinoContainer startTrinoNode(TrinoNodeSpec spec, Supplier<TrinoContainer.Builder> builder) {
823+
synchronized (SharedContainers.class) {
824+
return trinoNode.getOrStart(
825+
spec,
826+
() -> {
827+
initIfNecessary();
828+
TrinoContainer node = closer.register(builder.get().withNetwork(network).build());
829+
start(node);
830+
return node;
831+
});
832+
}
833+
}
834+
792835
HiveContainer getHiveContainer() {
793836
return hiveContainer.get();
794837
}

‎integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TestContainerExtension.java‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,7 @@ void testUnsupportedPrivateScopeFailsWithTheSpecsReason() {
209209

210210
@Test
211211
void testDurableFixtureWithoutPrivateContainersFailsWithTheSpecsReason() {
212-
run(DurableTrinoFixture.class)
212+
run(DurableClickHouseClusterFixture.class)
213213
.allEvents()
214214
.assertThatEvents()
215215
.haveAtLeast(
@@ -220,8 +220,9 @@ void testDurableFixtureWithoutPrivateContainersFailsWithTheSpecsReason() {
220220
message(
221221
m ->
222222
m.startsWith(
223-
"@TrinoFixture(durability = DURABLE) needs a container of its"
224-
+ " own, but TrinoSpec has no private clusters")))));
223+
"@ClickHouseFixture(durability = DURABLE) needs a container of"
224+
+ " its own, but ClickHouseSpec has no private three-node"
225+
+ " cluster")))));
225226
}
226227

227228
@Test
@@ -700,8 +701,11 @@ void test() {}
700701
}
701702

702703
@Tag(FIXTURE)
703-
@TrinoFixture(durability = Durability.DURABLE)
704-
static class DurableTrinoFixture {
704+
@ClickHouseFixture(
705+
nodes = 3,
706+
database = TestDatabaseName.CLICKHOUSE_CLUSTER_CLICKHOUSE_IT,
707+
durability = Durability.DURABLE)
708+
static class DurableClickHouseClusterFixture {
705709
@Test
706710
void test() {}
707711
}

‎integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TestPrivateScope.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -53,15 +53,15 @@ void testEveryFixtureReadsItsScope() throws Exception {
5353
}
5454

5555
@Test
56-
void testEverySpecButTrinoAndTheClickHouseClusterSupportsPrivate() throws Exception {
56+
void testEverySpecButTheClickHouseClusterSupportsPrivate() throws Exception {
5757
List<String> withoutPrivate = new ArrayList<>();
5858
for (Annotation fixture : fixtures(DefaultFixtures.class)) {
5959
ContainerSpec<?> spec = factory(fixture).specFor(fixture);
6060
if (!spec.supportsPrivate()) {
6161
withoutPrivate.add(spec.getClass().getSimpleName());
6262
}
6363
}
64-
Assertions.assertEquals(List.of("TrinoSpec"), withoutPrivate);
64+
Assertions.assertEquals(List.of(), withoutPrivate);
6565
Assertions.assertFalse(
6666
new ClickHouseSpec(3, TestDatabaseName.CLICKHOUSE_CLUSTER_CLICKHOUSE_IT).supportsPrivate());
6767
for (HiveStorage storage : HiveStorage.values()) {

‎integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TestTrinoCluster.java‎

Lines changed: 77 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,89 @@
1818
*/
1919
package org.apache.gravitino.integration.test.container;
2020

21+
import com.sun.net.httpserver.HttpExchange;
22+
import com.sun.net.httpserver.HttpServer;
23+
import java.io.IOException;
24+
import java.io.OutputStream;
25+
import java.net.InetAddress;
26+
import java.net.InetSocketAddress;
27+
import java.nio.charset.StandardCharsets;
28+
import java.util.ArrayList;
29+
import java.util.List;
2130
import java.util.Optional;
31+
import java.util.concurrent.atomic.AtomicInteger;
2232
import org.junit.jupiter.api.Assertions;
2333
import org.junit.jupiter.api.Test;
34+
import org.mockito.Mockito;
35+
import org.testcontainers.containers.wait.strategy.WaitStrategyTarget;
2436

25-
/** Tests how {@link TrinoCluster} reads Trino statement responses; nothing here starts Trino. */
37+
/**
38+
* Tests how {@link TrinoCluster} reads Trino responses and waits for a coordinator; a stub HTTP
39+
* server stands in for Trino, so nothing here starts a container.
40+
*/
2641
public class TestTrinoCluster {
2742

43+
@Test
44+
void testInfoSaysWhetherTheNodeHasStarted() {
45+
Assertions.assertFalse(TrinoCluster.hasStarted("{\"coordinator\":true,\"starting\":true}"));
46+
Assertions.assertTrue(
47+
TrinoCluster.hasStarted("{\"coordinator\":true, \"starting\" : false,\"uptime\":\"1s\"}"));
48+
Assertions.assertFalse(TrinoCluster.hasStarted(""));
49+
}
50+
51+
@Test
52+
void testCoordinatorIsReadyOnceStartedWithTheCatalogAndEveryWorker() throws IOException {
53+
AtomicInteger infoCalls = new AtomicInteger();
54+
List<String> queries = new ArrayList<>();
55+
HttpServer trino =
56+
HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
57+
// Still starting on the first call; then up with the catalog and two active workers.
58+
trino.createContext(
59+
"/v1/info",
60+
exchange -> respond(exchange, "{\"starting\":" + (infoCalls.incrementAndGet() == 1) + "}"));
61+
trino.createContext(
62+
"/v1/statement",
63+
exchange -> {
64+
String sql = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
65+
queries.add(sql);
66+
int count = sql.contains("system.runtime.nodes") ? 2 : 1;
67+
respond(exchange, "{\"id\":\"q\",\"data\":[[" + count + "]]}");
68+
});
69+
trino.start();
70+
try {
71+
WaitStrategyTarget target = Mockito.mock(WaitStrategyTarget.class);
72+
Mockito.when(target.getHost()).thenReturn(InetAddress.getLoopbackAddress().getHostAddress());
73+
Mockito.when(target.getMappedPort(TrinoCluster.TRINO_PORT))
74+
.thenReturn(trino.getAddress().getPort());
75+
76+
TrinoCluster.coordinatorReadiness(2).waitUntilReady(target);
77+
78+
Assertions.assertEquals(2, infoCalls.get());
79+
Assertions.assertEquals(2, queries.size(), queries.toString());
80+
Assertions.assertTrue(queries.get(0).contains("catalog_name = 'gravitino'"), queries.get(0));
81+
Assertions.assertTrue(queries.get(1).contains("state = 'active'"), queries.get(1));
82+
} finally {
83+
trino.stop(0);
84+
}
85+
}
86+
87+
@Test
88+
void testPrivateClusterGivesItsCoordinatorARunName() {
89+
Assertions.assertEquals("trino", TrinoCluster.builder().coordinatorHost());
90+
String name = TrinoCluster.builder().withPrivateNames(true).coordinatorHost();
91+
Assertions.assertTrue(name.startsWith("gravitino-trino-" + RunNames.runId() + "-"), name);
92+
Assertions.assertNotEquals(
93+
name, TrinoCluster.builder().withPrivateNames(true).coordinatorHost());
94+
}
95+
96+
private static void respond(HttpExchange exchange, String body) throws IOException {
97+
byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
98+
exchange.sendResponseHeaders(200, bytes.length);
99+
try (OutputStream out = exchange.getResponseBody()) {
100+
out.write(bytes);
101+
}
102+
}
103+
28104
@Test
29105
void testQueuedPageHasNextUriAndNoValue() {
30106
String queued =

‎integration-test-common/src/test/java/org/apache/gravitino/integration/test/container/TestTrinoSpec.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,14 @@
1717
* under the License.
1818
*/ package org.apache.gravitino.integration.test.container;
1919

20+
import java.io.IOException;
21+
import java.nio.file.Files;
22+
import java.nio.file.Path;
2023
import java.util.List;
24+
import java.util.Map;
2125
import org.junit.jupiter.api.Assertions;
2226
import org.junit.jupiter.api.Test;
27+
import org.junit.jupiter.api.io.TempDir;
2328

2429
public class TestTrinoSpec {
2530

@@ -51,6 +56,56 @@ void testTrinoFixtureStartsLazilyByDefault() throws NoSuchMethodException {
5156
Assertions.assertEquals(List.of(), new TrinoSpec(478, 0, HiveVersion.HIVE2, 1).dependencies());
5257
}
5358

59+
@Test
60+
void testPrivateClusterNeedsAServerToo() {
61+
TrinoSpec unbound = new TrinoSpec(478, 1, HiveVersion.HIVE2, -1);
62+
63+
Assertions.assertTrue(unbound.supportsPrivate());
64+
IllegalStateException e =
65+
Assertions.assertThrows(
66+
IllegalStateException.class, () -> unbound.startPrivate(null, Durability.DISPOSABLE));
67+
Assertions.assertTrue(e.getMessage().contains("GravitinoServerAware"), e.getMessage());
68+
}
69+
70+
@Test
71+
void testNodeDependsOnTheSharedHive2AndNeedsAServer() {
72+
TrinoNodeSpec node = new TrinoNodeSpec("lake", 8090);
73+
74+
Assertions.assertEquals(
75+
List.of(new HiveSpec(HiveVersion.HIVE2, HiveAuth.NONE)), node.dependencies());
76+
Assertions.assertEquals(TrinoContainer.class, node.containerType());
77+
Assertions.assertTrue(node.supportsPrivate());
78+
Assertions.assertThrows(IllegalArgumentException.class, () -> new TrinoNodeSpec("lake", -1));
79+
}
80+
81+
@Test
82+
void testNodeIsBuiltForItsServerMetalakeAndHive() {
83+
TrinoContainer node =
84+
new TrinoNodeSpec("lake", 8090).builder("10.0.0.5", "/connector", null).build();
85+
86+
Assertions.assertEquals(TestImages.TRINO, node.getContainer().getDockerImageName());
87+
Assertions.assertEquals("8090", node.getContainer().getEnvMap().get("GRAVITINO_HOST_PORT"));
88+
Assertions.assertEquals("lake", node.getContainer().getEnvMap().get("GRAVITINO_METALAKE_NAME"));
89+
Assertions.assertTrue(
90+
node.getContainer().getExtraHosts().contains(HiveContainer.HOST_NAME + ":10.0.0.5"),
91+
node.getContainer().getExtraHosts().toString());
92+
Assertions.assertEquals(
93+
List.of(TrinoContainer.TRINO_PORT), node.getContainer().getExposedPorts());
94+
}
95+
96+
@Test
97+
void testNodePluginFilesAddEveryExtraJar(@TempDir Path extra) throws IOException {
98+
Files.createFile(extra.resolve("a.jar"));
99+
Files.createFile(extra.resolve("notes.txt"));
100+
String plugin = TrinoContainer.TRINO_CONTAINER_PLUGIN_GRAVITINO_DIR;
101+
102+
Assertions.assertEquals(
103+
Map.of(plugin, "/connector"), TrinoNodeSpec.pluginFiles("/connector", null));
104+
Assertions.assertEquals(
105+
Map.of(plugin, "/connector", plugin + "/a.jar", extra.resolve("a.jar").toString()),
106+
TrinoNodeSpec.pluginFiles("/connector", extra.toString()));
107+
}
108+
54109
@TrinoFixture(workers = 2, hive = HiveVersion.HIVE3)
55110
static class TrinoOnly {}
56111
}

0 commit comments

Comments
 (0)