diff --git a/.github/actions/java-gradle/pre-merge/action.yml b/.github/actions/java-gradle/pre-merge/action.yml index 565e33c30f..d5568883b5 100644 --- a/.github/actions/java-gradle/pre-merge/action.yml +++ b/.github/actions/java-gradle/pre-merge/action.yml @@ -143,6 +143,11 @@ runs: mkdir -p reports/java-tests cp -r foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/build/test-results reports/java-tests/flink fi + if [ -d "foreign/java/external-processors/iggy-connector-pinot/build/test-results" ]; then + echo "Found test reports in pinot" + mkdir -p reports/java-tests + cp -r foreign/java/external-processors/iggy-connector-pinot/build/test-results reports/java-tests/pinot + fi - name: Stop Iggy server if: always() && inputs.task == 'test' @@ -188,4 +193,5 @@ runs: paths: | foreign/java/java-sdk/build/test-results/**/TEST-*.xml foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/build/test-results/**/TEST-*.xml + foreign/java/external-processors/iggy-connector-pinot/build/test-results/**/TEST-*.xml if: ${{ !cancelled() && inputs.task == 'test' }} diff --git a/foreign/java/external-processors/iggy-connector-pinot/build.gradle.kts b/foreign/java/external-processors/iggy-connector-pinot/build.gradle.kts index 769d1877b7..88d5e8bec4 100644 --- a/foreign/java/external-processors/iggy-connector-pinot/build.gradle.kts +++ b/foreign/java/external-processors/iggy-connector-pinot/build.gradle.kts @@ -19,6 +19,7 @@ plugins { id("iggy.java-library-conventions") + alias(libs.plugins.shadow) } dependencies { @@ -43,13 +44,19 @@ dependencies { testImplementation(platform(libs.junit.bom)) testImplementation(libs.bundles.testing) testImplementation(libs.pinot.spi) // Need Pinot SPI for tests + testImplementation(libs.testcontainers) testRuntimeOnly(libs.slf4j.simple) } -// Assemble connector plugin with all dependencies for Docker deployment -tasks.register("assemblePlugin") { - from(tasks.named("jar")) - from(configurations.runtimeClasspath) +tasks.shadowJar { + duplicatesStrategy = DuplicatesStrategy.EXCLUDE + relocate("io.netty", "org.apache.iggy.connector.pinot.shaded.io.netty") + mergeServiceFiles() +} + +// Assemble connector plugin with isolated dependencies for Docker deployment +tasks.register("assemblePlugin") { + from(tasks.named("shadowJar")) into(layout.buildDirectory.dir("plugin")) } @@ -57,6 +64,16 @@ tasks.named("jar") { finalizedBy("assemblePlugin") } +tasks.named("test") { + dependsOn("assemblePlugin") + inputs.dir(layout.projectDirectory.dir("deployment")) + inputs.property("useExternalServer", providers.environmentVariable("USE_EXTERNAL_SERVER").isPresent) + inputs.property("externalTcpPort", providers.environmentVariable("EXTERNAL_TCP_PORT").orElse("8090")) + systemProperty("iggy.pinot.image", "apachepinot/pinot:${libs.versions.pinot.get()}") + systemProperty("iggy.pinot.plugin.dir", layout.buildDirectory.dir("plugin").get().asFile.absolutePath) + systemProperty("iggy.pinot.deployment.dir", layout.projectDirectory.dir("deployment").asFile.absolutePath) +} + publishing { publications { named("maven") { diff --git a/foreign/java/external-processors/iggy-connector-pinot/integration-test.sh b/foreign/java/external-processors/iggy-connector-pinot/integration-test.sh deleted file mode 100755 index f83c5f8ec1..0000000000 --- a/foreign/java/external-processors/iggy-connector-pinot/integration-test.sh +++ /dev/null @@ -1,234 +0,0 @@ -#!/usr/bin/env bash -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you 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. - -set -e - -# Colors for output -GREEN='\033[0;32m' -YELLOW='\033[1;33m' -RED='\033[0;31m' -NC='\033[0m' # No Color - -echo -e "${GREEN}=====================================${NC}" -echo -e "${GREEN}Iggy-Pinot Integration Test${NC}" -echo -e "${GREEN}=====================================${NC}" - -# Navigate to connector directory -cd "$(dirname "$0")" - -# Step 1: Build JARs -echo -e "\n${YELLOW}Step 1: Building JARs...${NC}" -cd ../../ -gradle :iggy-connector-pinot:jar :iggy:jar -cd external-processors/iggy-connector-pinot -echo -e "${GREEN}✓ JARs built successfully${NC}" - -# Step 2: Start Docker environment -echo -e "\n${YELLOW}Step 2: Starting Docker environment...${NC}" -docker-compose down -v -docker-compose up -d -echo -e "${GREEN}✓ Docker containers starting${NC}" - -# Step 3: Wait for services to be healthy -echo -e "\n${YELLOW}Step 3: Waiting for services to be healthy...${NC}" - -echo -n "Waiting for Iggy... " -for i in {1..30}; do - if curl --connect-timeout 3 --max-time 5 -s http://localhost:3000/ > /dev/null 2>&1; then - echo -e "${GREEN}✓${NC}" - break - fi - sleep 2 - echo -n "." -done - -echo -n "Waiting for Pinot Controller... " -for i in {1..60}; do - if curl --connect-timeout 3 --max-time 5 -s http://localhost:9000/health > /dev/null 2>&1; then - echo -e "${GREEN}✓${NC}" - break - fi - sleep 2 - echo -n "." -done - -echo -n "Waiting for Pinot Broker... " -for i in {1..60}; do - if curl --connect-timeout 3 --max-time 5 -s http://localhost:8099/health > /dev/null 2>&1; then - echo -e "${GREEN}✓${NC}" - break - fi - sleep 2 - echo -n "." -done - -echo -n "Waiting for Pinot Server... " -for i in {1..60}; do - if curl --connect-timeout 3 --max-time 5 -s http://localhost:8097/health > /dev/null 2>&1; then - echo -e "${GREEN}✓${NC}" - break - fi - sleep 2 - echo -n "." -done - -sleep 5 # Extra time for services to stabilize - -# Step 4: Login to Iggy and create stream/topic -echo -e "\n${YELLOW}Step 4: Logging in to Iggy and creating stream/topic...${NC}" - -# Login and get JWT token -TOKEN=$(curl -s -X POST "http://localhost:3000/users/login" \ - -H "Content-Type: application/json" \ - -d '{"username": "iggy", "password": "iggy"}' | jq -r '.access_token.token') - -if [ -z "$TOKEN" ] || [ "$TOKEN" = "null" ]; then - echo -e "${RED}✗ Failed to get authentication token${NC}" - exit 1 -fi - -echo -e "${GREEN}✓ Authenticated${NC}" - -# Create stream -curl -s -X POST "http://localhost:3000/streams" \ - -H "Authorization: Bearer $TOKEN" \ - -H "Content-Type: application/json" \ - -d '{"stream_id": 1, "name": "test-stream"}' \ - && echo -e "${GREEN}✓ Stream created${NC}" || echo -e "${RED}✗ Stream creation failed (may already exist)${NC}" - -# Create topic -TOPIC_RESPONSE=$(curl -s -X POST "http://localhost:3000/streams/test-stream/topics" \ - -H "Authorization: Bearer $TOKEN" \ - -H "Content-Type: application/json" \ - -d '{"topic_id": 1, "name": "test-events", "partitions_count": 2, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 0}') - -if echo "$TOPIC_RESPONSE" | grep -q '"id"'; then - echo -e "${GREEN}✓ Topic created${NC}" -else - echo -e "${RED}✗ Topic creation failed: $TOPIC_RESPONSE${NC}" - exit 1 -fi - -# Create consumer group (topic-scoped, not stream-scoped) -curl -s -X POST "http://localhost:3000/streams/test-stream/topics/test-events/consumer-groups" \ - -H "Authorization: Bearer $TOKEN" \ - -H "Content-Type: application/json" \ - -d '{"name": "pinot-integration-test"}' \ - && echo -e "${GREEN}✓ Consumer group created${NC}" || echo -e "${YELLOW}Note: Consumer group may already exist${NC}" - -# Step 5: Create Pinot schema -echo -e "\n${YELLOW}Step 5: Creating Pinot schema...${NC}" -curl -X POST "http://localhost:9000/schemas" \ - -H "Content-Type: application/json" \ - -d @deployment/schema.json \ - && echo -e "${GREEN}✓ Schema created${NC}" || echo -e "${RED}✗ Schema creation failed${NC}" - -# Step 6: Create Pinot table -echo -e "\n${YELLOW}Step 6: Creating Pinot realtime table...${NC}" -TABLE_RESPONSE=$(curl -s -X POST "http://localhost:9000/tables" \ - -H "Content-Type: application/json" \ - -d @deployment/table.json) - -if echo "$TABLE_RESPONSE" | grep -q '"status":"Table test_events_REALTIME successfully added"'; then - echo -e "${GREEN}✓ Table created${NC}" -elif echo "$TABLE_RESPONSE" | grep -q '"code":500'; then - echo -e "${RED}✗ Table creation failed${NC}" - echo "$TABLE_RESPONSE" | jq '.' - exit 1 -else - echo -e "${GREEN}✓ Table created${NC}" -fi - -sleep 5 # Let table initialize - -# Step 7: Send test messages to Iggy -echo -e "\n${YELLOW}Step 7: Sending test messages to Iggy...${NC}" - -# Partition value for partition 0 (4-byte little-endian, base64 encoded) -PARTITION_VALUE=$(printf '\x00\x00\x00\x00' | base64) - -for i in {1..10}; do - TIMESTAMP=$(($(date +%s) * 1000)) - MESSAGE=$(cat < /dev/null 2>&1 - echo -e "${GREEN}✓ Message $i sent${NC}" - sleep 1 -done - -# Step 8: Wait for ingestion -echo -e "\n${YELLOW}Step 8: Waiting for Pinot to ingest messages...${NC}" -sleep 15 - -# Step 9: Query Pinot and verify data -echo -e "\n${YELLOW}Step 9: Querying Pinot for ingested data...${NC}" - -QUERY_RESULT=$(curl -s -X POST "http://localhost:8099/query/sql" \ - -H "Content-Type: application/json" \ - -d '{"sql": "SELECT COUNT(*) FROM test_events_REALTIME"}') - -echo "Query Result:" -echo "$QUERY_RESULT" | jq '.' - -# Extract count from result -COUNT=$(echo "$QUERY_RESULT" | jq -r '.resultTable.rows[0][0]' 2>/dev/null || echo "0") - -if [ "$COUNT" -gt "0" ]; then - echo -e "\n${GREEN}=====================================${NC}" - echo -e "${GREEN}✓ Integration Test PASSED!${NC}" - echo -e "${GREEN}Successfully ingested $COUNT messages${NC}" - echo -e "${GREEN}=====================================${NC}" - - # Show sample data - echo -e "\n${YELLOW}Sample data:${NC}" - curl -s -X POST "http://localhost:8099/query/sql" \ - -H "Content-Type: application/json" \ - -d '{"sql": "SELECT * FROM test_events_REALTIME LIMIT 5"}' | jq '.' - - EXIT_CODE=0 -else - echo -e "\n${RED}=====================================${NC}" - echo -e "${RED}✗ Integration Test FAILED!${NC}" - echo -e "${RED}No messages ingested${NC}" - echo -e "${RED}=====================================${NC}" - - # Show logs for debugging - echo -e "\n${YELLOW}Pinot Server logs:${NC}" - docker logs pinot-server --tail 50 - - EXIT_CODE=1 -fi - -# Cleanup option -echo -e "\n${YELLOW}To stop the environment: docker-compose down -v${NC}" -echo -e "${YELLOW}To view logs: docker-compose logs -f${NC}" - -exit $EXIT_CODE diff --git a/foreign/java/external-processors/iggy-connector-pinot/src/test/java/org/apache/iggy/connector/pinot/IggyPinotIntegrationTest.java b/foreign/java/external-processors/iggy-connector-pinot/src/test/java/org/apache/iggy/connector/pinot/IggyPinotIntegrationTest.java new file mode 100644 index 0000000000..5d78873da0 --- /dev/null +++ b/foreign/java/external-processors/iggy-connector-pinot/src/test/java/org/apache/iggy/connector/pinot/IggyPinotIntegrationTest.java @@ -0,0 +1,465 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.iggy.connector.pinot; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.github.dockerjava.api.model.Capability; +import com.github.dockerjava.api.model.Ulimit; +import org.apache.iggy.client.blocking.tcp.IggyTcpClient; +import org.apache.iggy.identifier.StreamId; +import org.apache.iggy.identifier.TopicId; +import org.apache.iggy.message.Message; +import org.apache.iggy.message.Partitioning; +import org.apache.iggy.topic.CompressionAlgorithm; +import org.apache.pinot.spi.exception.QueryErrorCode; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.testcontainers.Testcontainers; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.PullPolicy; +import org.testcontainers.utility.DockerImageName; +import org.testcontainers.utility.MountableFile; + +import java.io.IOException; +import java.math.BigInteger; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.UUID; +import java.util.function.Predicate; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + +class IggyPinotIntegrationTest { + + // The Java SDK speaks VSR, so use the same VSR-capable image as its integration tests. + private static final DockerImageName IGGY_IMAGE = DockerImageName.parse("apache/iggy:edge"); + private static final DockerImageName PINOT_IMAGE = DockerImageName.parse( + Objects.requireNonNull(System.getProperty("iggy.pinot.image"), "Missing iggy.pinot.image system property")); + private static final DockerImageName ZOOKEEPER_IMAGE = DockerImageName.parse("zookeeper:3.9"); + + private static final int IGGY_HTTP_PORT = 3000; + private static final int IGGY_TCP_PORT = 8090; + private static final int PINOT_CONTROLLER_PORT = 9000; + private static final int PINOT_BROKER_PORT = 8099; + private static final int PINOT_SERVER_ADMIN_PORT = 8097; + private static final String EXTERNAL_SERVER_HOST = "127.0.0.1"; + private static final String TESTCONTAINERS_HOST = "host.testcontainers.internal"; + private static final boolean USE_EXTERNAL_SERVER = System.getenv("USE_EXTERNAL_SERVER") != null; + + private static final String STREAM_NAME = "pinot-test-stream-" + UUID.randomUUID(); + private static final String TOPIC_NAME = "test-events"; + private static final String CONSUMER_GROUP_NAME = "pinot-integration-test"; + private static final String TABLE_NAME = "test_events_REALTIME"; + + private static final Duration HTTP_TIMEOUT = Duration.ofSeconds(30); + private static final Duration QUERY_TIMEOUT = Duration.ofSeconds(90); + private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(3); + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final HttpClient HTTP_CLIENT = HttpClient.newBuilder() + .connectTimeout(HTTP_TIMEOUT) + .version(HttpClient.Version.HTTP_1_1) + .build(); + + private static Network network; + private static GenericContainer iggy; + private static GenericContainer zookeeper; + private static GenericContainer pinotController; + private static GenericContainer pinotBroker; + private static GenericContainer pinotServer; + private static IggyTcpClient iggyClient; + + @BeforeAll + static void startEnvironment() { + Path pluginDirectory = requiredDirectory("iggy.pinot.plugin.dir"); + Path deploymentDirectory = requiredDirectory("iggy.pinot.deployment.dir"); + + if (USE_EXTERNAL_SERVER) { + Testcontainers.exposeHostPorts(externalTcpPort()); + } + network = Network.newNetwork(); + try { + startZookeeper(); + if (!USE_EXTERNAL_SERVER) { + startIggy(); + } + startPinotController(pluginDirectory); + startPinotBroker(); + startPinotServer(pluginDirectory); + + iggyClient = IggyTcpClient.builder() + .host(iggyHost()) + .port(iggyPort()) + .credentials("iggy", "iggy") + .connectionTimeout(Duration.ofSeconds(10)) + .requestTimeout(Duration.ofSeconds(10)) + .buildAndLogin(); + createIggyResources(); + + postControllerResource("/schemas", Files.readString(deploymentDirectory.resolve("schema.json"))); + postControllerResource("/tables", tableConfiguration(deploymentDirectory.resolve("table.json"))); + awaitQuery("SELECT COUNT(*) FROM " + TABLE_NAME, IggyPinotIntegrationTest::hasResultRow); + } catch (IOException | RuntimeException e) { + throw new IllegalStateException("Failed to start the Iggy-Pinot test environment\n" + diagnostics(), e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while starting the Iggy-Pinot test environment", e); + } + } + + @AfterAll + static void stopEnvironment() { + stop(pinotServer); + stop(pinotBroker); + stop(pinotController); + + if (iggyClient != null) { + try { + iggyClient.streams().deleteStream(StreamId.of(STREAM_NAME)); + } catch (RuntimeException ignored) { + // Startup may have failed before the stream was created. + } + try { + iggyClient.close(); + } catch (RuntimeException ignored) { + // Containers are still stopped below. + } + } + + stop(iggy); + stop(zookeeper); + if (network != null) { + network.close(); + } + } + + @Test + void shouldIngestAndMapJsonMessage() throws Exception { + String marker = "mapping-" + UUID.randomUUID(); + long timestamp = Instant.now().toEpochMilli(); + String payload = jsonMessage(marker, "account-updated", "mobile", 750L, timestamp); + + sendMessages(List.of(Message.of(payload))); + + JsonNode result = awaitQuery( + "SELECT * FROM " + TABLE_NAME + " WHERE userId = '" + marker + "' LIMIT 1", + IggyPinotIntegrationTest::hasResultRow); + + assertThat(value(result, "userId").asText()).isEqualTo(marker); + assertThat(value(result, "eventType").asText()).isEqualTo("account-updated"); + assertThat(value(result, "deviceType").asText()).isEqualTo("mobile"); + assertThat(value(result, "duration").asLong()).isEqualTo(750L); + assertThat(value(result, "timestamp").asLong()).isEqualTo(timestamp); + } + + @Test + void shouldIngestMessageBatch() throws Exception { + String marker = "batch-" + UUID.randomUUID(); + int batchSize = 10; + List messages = new ArrayList<>(batchSize); + for (int i = 0; i < batchSize; i++) { + messages.add(Message.of(jsonMessage( + marker + "-" + i, + marker, + i % 2 == 0 ? "desktop" : "mobile", + i * 100L, + Instant.now().toEpochMilli() + i))); + } + + sendMessages(messages); + + JsonNode result = awaitQuery( + "SELECT COUNT(*) FROM " + TABLE_NAME + " WHERE eventType = '" + marker + "'", + response -> firstValue(response).asInt() == batchSize); + + assertThat(firstValue(result).asInt()).isEqualTo(batchSize); + } + + private static void startZookeeper() { + zookeeper = new GenericContainer<>(ZOOKEEPER_IMAGE) + .withNetwork(network) + .withNetworkAliases("zookeeper") + .withExposedPorts(2181) + .withEnv("ZOOKEEPER_CLIENT_PORT", "2181") + .withEnv("ZOOKEEPER_TICK_TIME", "2000") + .waitingFor(Wait.forListeningPort().withStartupTimeout(STARTUP_TIMEOUT)); + zookeeper.start(); + } + + private static void startIggy() { + iggy = new GenericContainer<>(IGGY_IMAGE) + .withImagePullPolicy(PullPolicy.alwaysPull()) + .withNetwork(network) + .withNetworkAliases("iggy") + .withExposedPorts(IGGY_HTTP_PORT, IGGY_TCP_PORT) + .withEnv("IGGY_SYSTEM_LOGGING_LEVEL", "info") + .withEnv("IGGY_TCP_ADDRESS", "0.0.0.0:8090") + .withEnv("IGGY_HTTP_ENABLED", "true") + .withEnv("IGGY_HTTP_ADDRESS", "0.0.0.0:3000") + .withEnv("IGGY_ROOT_USERNAME", "iggy") + .withEnv("IGGY_ROOT_PASSWORD", "iggy") + .withEnv("IGGY_SYSTEM_SHARDING_CPU_ALLOCATION", "1") + .withCreateContainerCmdModifier(cmd -> cmd.getHostConfig() + .withCapAdd(Capability.SYS_NICE) + .withSecurityOpts(List.of("seccomp:unconfined")) + .withUlimits(List.of(new Ulimit("memlock", -1L, -1L)))) + .waitingFor(Wait.forHttp("/") + .forPort(IGGY_HTTP_PORT) + .forStatusCodeMatching(status -> status >= 200 && status < 500) + .withStartupTimeout(STARTUP_TIMEOUT)); + iggy.start(); + } + + private static String iggyHost() { + return USE_EXTERNAL_SERVER ? EXTERNAL_SERVER_HOST : iggy.getHost(); + } + + private static int iggyPort() { + return USE_EXTERNAL_SERVER ? externalTcpPort() : iggy.getMappedPort(IGGY_TCP_PORT); + } + + private static int externalTcpPort() { + String configured = System.getenv("EXTERNAL_TCP_PORT"); + return configured != null ? Integer.parseInt(configured) : IGGY_TCP_PORT; + } + + private static void startPinotController(Path pluginDirectory) { + pinotController = pinotContainer(pluginDirectory) + .withNetworkAliases("pinot-controller") + .withExposedPorts(PINOT_CONTROLLER_PORT) + .withCommand("StartController", "-zkAddress", "zookeeper:2181") + .withEnv("JAVA_OPTS", "-Xms512M -Xmx1G -XX:+UseG1GC -Dplugins.include=iggy-connector") + .waitingFor( + Wait.forHttp("/health").forPort(PINOT_CONTROLLER_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + pinotController.start(); + } + + private static void startPinotBroker() { + pinotBroker = new GenericContainer<>(PINOT_IMAGE) + .withNetwork(network) + .withNetworkAliases("pinot-broker") + .withExposedPorts(PINOT_BROKER_PORT) + .withCommand("StartBroker", "-zkAddress", "zookeeper:2181") + .withEnv("JAVA_OPTS", "-Xms512M -Xmx1G -XX:+UseG1GC") + .waitingFor(Wait.forHttp("/health").forPort(PINOT_BROKER_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + pinotBroker.start(); + } + + private static void startPinotServer(Path pluginDirectory) { + pinotServer = pinotContainer(pluginDirectory) + .withNetworkAliases("pinot-server") + .withExposedPorts(PINOT_SERVER_ADMIN_PORT) + .withCommand("StartServer", "-zkAddress", "zookeeper:2181") + .withEnv("JAVA_OPTS", "-Xms512M -Xmx1G -XX:+UseG1GC -Dplugins.include=iggy-connector") + .waitingFor( + Wait.forHttp("/health").forPort(PINOT_SERVER_ADMIN_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + pinotServer.start(); + } + + private static GenericContainer pinotContainer(Path pluginDirectory) { + return new GenericContainer<>(PINOT_IMAGE) + .withNetwork(network) + .withCopyFileToContainer( + MountableFile.forHostPath(pluginDirectory), + "/opt/pinot/plugins/pinot-stream-ingestion/iggy-connector"); + } + + private static void createIggyResources() { + iggyClient.streams().createStream(STREAM_NAME); + StreamId streamId = StreamId.of(STREAM_NAME); + iggyClient + .topics() + .createTopic(streamId, 2L, CompressionAlgorithm.None, BigInteger.ZERO, BigInteger.ZERO, TOPIC_NAME); + iggyClient.consumerGroups().createConsumerGroup(streamId, TopicId.of(TOPIC_NAME), CONSUMER_GROUP_NAME); + } + + private static void sendMessages(List messages) { + iggyClient + .messages() + .sendMessages(StreamId.of(STREAM_NAME), TopicId.of(TOPIC_NAME), Partitioning.partitionId(0L), messages); + } + + private static String jsonMessage(String userId, String eventType, String deviceType, long duration, long timestamp) + throws IOException { + return OBJECT_MAPPER.writeValueAsString(OBJECT_MAPPER + .createObjectNode() + .put("userId", userId) + .put("eventType", eventType) + .put("deviceType", deviceType) + .put("duration", duration) + .put("timestamp", timestamp)); + } + + private static String tableConfiguration(Path tableConfigurationPath) throws IOException { + JsonNode tableConfiguration = OBJECT_MAPPER.readTree(Files.readString(tableConfigurationPath)); + ObjectNode streamConfigs = + (ObjectNode) tableConfiguration.required("tableIndexConfig").required("streamConfigs"); + streamConfigs.put("stream.iggy.stream.id", STREAM_NAME); + if (USE_EXTERNAL_SERVER) { + streamConfigs.put("stream.iggy.host", TESTCONTAINERS_HOST); + streamConfigs.put("stream.iggy.port", Integer.toString(externalTcpPort())); + } + return OBJECT_MAPPER.writeValueAsString(tableConfiguration); + } + + private static void postControllerResource(String path, String body) throws IOException, InterruptedException { + HttpResponse response = post(pinotController, PINOT_CONTROLLER_PORT, path, body); + if (response.statusCode() < 200 || response.statusCode() >= 300) { + throw new IllegalStateException("Pinot controller request to %s failed with status %d: %s" + .formatted(path, response.statusCode(), response.body())); + } + } + + private static JsonNode awaitQuery(String sql, Predicate success) { + long deadline = System.nanoTime() + QUERY_TIMEOUT.toNanos(); + String lastResponse = "No response received"; + + while (System.nanoTime() < deadline) { + try { + HttpResponse response = post( + pinotBroker, + PINOT_BROKER_PORT, + "/query/sql", + OBJECT_MAPPER.createObjectNode().put("sql", sql).toString()); + lastResponse = "HTTP " + response.statusCode() + ": " + response.body(); + if (response.statusCode() >= 200 && response.statusCode() < 300) { + JsonNode json = OBJECT_MAPPER.readTree(response.body()); + if (hasExceptionCode(json, QueryErrorCode.SQL_PARSING.getId())) { + return fail("Pinot rejected SQL query: %s%nResponse: %s".formatted(sql, response.body())); + } + if (json.path("exceptions").isEmpty() && success.test(json)) { + return json; + } + } + } catch (IOException e) { + lastResponse = e.toString(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while waiting for Pinot query", e); + } + + try { + Thread.sleep(500); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Interrupted while waiting for Pinot query", e); + } + } + + return fail("Pinot query did not reach the expected result within %s.%nSQL: %s%nLast response: %s%n%s" + .formatted(QUERY_TIMEOUT, sql, lastResponse, diagnostics())); + } + + private static HttpResponse post(GenericContainer container, int port, String path, String body) + throws IOException, InterruptedException { + URI uri = URI.create("http://" + container.getHost() + ":" + container.getMappedPort(port) + path); + HttpRequest request = HttpRequest.newBuilder(uri) + .timeout(HTTP_TIMEOUT) + .header("Content-Type", "application/json") + .POST(HttpRequest.BodyPublishers.ofString(body)) + .build(); + return HTTP_CLIENT.send(request, HttpResponse.BodyHandlers.ofString()); + } + + private static boolean hasResultRow(JsonNode response) { + return response.path("resultTable").path("rows").size() > 0; + } + + private static JsonNode firstValue(JsonNode response) { + return response.path("resultTable").path("rows").path(0).path(0); + } + + private static JsonNode value(JsonNode response, String columnName) { + JsonNode columnNames = response.path("resultTable").path("dataSchema").path("columnNames"); + for (int i = 0; i < columnNames.size(); i++) { + if (columnName.equals(columnNames.get(i).asText())) { + return response.path("resultTable").path("rows").path(0).path(i); + } + } + return fail("Pinot result did not contain column '%s': %s".formatted(columnName, response)); + } + + private static boolean hasExceptionCode(JsonNode response, int errorCode) { + for (JsonNode exception : response.path("exceptions")) { + if (exception.path("errorCode").asInt() == errorCode) { + return true; + } + } + return false; + } + + private static Path requiredDirectory(String property) { + String value = System.getProperty(property); + if (value == null) { + throw new IllegalStateException("Missing required system property: " + property); + } + Path directory = Path.of(value); + if (!Files.isDirectory(directory)) { + throw new IllegalStateException("Required directory does not exist: " + directory); + } + return directory; + } + + private static void stop(GenericContainer container) { + if (container != null) { + container.stop(); + } + } + + private static String diagnostics() { + return String.join( + "\n", + iggyDiagnostics(), + logs("Pinot controller", pinotController), + logs("Pinot broker", pinotBroker), + logs("Pinot server", pinotServer)); + } + + private static String iggyDiagnostics() { + if (USE_EXTERNAL_SERVER) { + return "Iggy uses external server at " + iggyHost() + ":" + iggyPort() + "."; + } + return logs("Iggy", iggy); + } + + private static String logs(String name, GenericContainer container) { + if (container == null || !container.isCreated()) { + return name + " was not created."; + } + String logs = container.getLogs(); + int start = Math.max(0, logs.length() - 4_000); + return "=== " + name + " logs ===\n" + logs.substring(start); + } +}