This is an automated email from the ASF dual-hosted git repository. ndipiazza pushed a commit to branch TIKA-4606-e2e-tests in repository https://gitbox.apache.org/repos/asf/tika.git
commit b24561eb8729842cbf19a0bd954b8cd2f0c5cc8d Author: Nicholas DiPiazza <[email protected]> AuthorDate: Wed Feb 25 18:16:20 2026 -0600 TIKA-4606: Add e2e tests for Ignite 3.x upgrade - Add tika-e2e-tests module to root pom.xml build - Update tika-e2e-tests parent pom: inherit from tika-parent, add Micronaut version alignment for Ignite 3.x transitive dependencies - Update tika-grpc e2e pom: local server mode by default (tika.e2e.useLocalServer=true), add Awaitility dependency, disable enforcer for Ignite 3.x dep conflicts - Add tika-config-ignite-local.json for local server mode testing - Update ExternalTestBase and IgniteConfigStoreTest for local vs Docker switching - Add FileSystemFetcherTest e2e test Note: these tests require the Ignite 3.x core upgrade from TIKA-4606-ignite-core. Co-authored-by: Copilot <[email protected]> --- pom.xml | 21 +- tika-e2e-tests/README.md | 6 +- tika-e2e-tests/pom.xml | 39 +- tika-e2e-tests/tika-grpc/README.md | 20 +- tika-e2e-tests/tika-grpc/pom.xml | 44 +- .../sample-configs/ignite/tika-config-ignite.json | 2 +- .../org/apache/tika/pipes/ExternalTestBase.java | 211 +++++++- .../pipes/filesystem/FileSystemFetcherTest.java | 50 +- .../tika/pipes/ignite/IgniteConfigStoreTest.java | 576 ++++++++++++++++----- ...g-ignite.json => tika-config-ignite-local.json} | 4 +- .../src/test/resources/tika-config-ignite.json | 2 +- 11 files changed, 783 insertions(+), 192 deletions(-) diff --git a/pom.xml b/pom.xml index e921bfb549..1511cb0954 100644 --- a/pom.xml +++ b/pom.xml @@ -57,11 +57,15 @@ <module>tika-example</module> <module>tika-java7</module> <module>tika-handlers</module> + <module>tika-e2e-tests</module> </modules> <profiles> <profile> <id>apache-release</id> + <modules> + <module>docs</module> + </modules> <properties> <username>${user.name}</username> </properties> @@ -189,16 +193,19 @@ least three +1 Tika PMC votes are cast. <groupId>org.apache.rat</groupId> <artifactId>apache-rat-plugin</artifactId> <configuration> - <excludes> - <exclude>CHANGES.txt</exclude> - <exclude>README.md</exclude> + <inputExcludes> + <inputExclude>CHANGES.txt</inputExclude> + <inputExclude>README.md</inputExclude> <!-- remove this once we figure out the bundle packaging --> - <exclude>tika-bundle/src/main/resources/META-INF/MANIFEST.MF</exclude> - <exclude>.gitattributes</exclude> + <inputExclude>tika-bundle/src/main/resources/META-INF/MANIFEST.MF</inputExclude> + <inputExclude>.gitattributes</inputExclude> + + <!-- Antora UI supplemental files --> + <inputExclude>docs/supplemental-ui/**</inputExclude> <!-- subprojects already checked, added for RAT 0.17, see also RAT-97 --> - <exclude>tika-*/**</exclude> - </excludes> + <inputExclude>tika-*/**</inputExclude> + </inputExcludes> </configuration> </plugin> </plugins> diff --git a/tika-e2e-tests/README.md b/tika-e2e-tests/README.md index 8c419571ae..163b0382aa 100644 --- a/tika-e2e-tests/README.md +++ b/tika-e2e-tests/README.md @@ -24,20 +24,20 @@ This module contains standalone end-to-end (E2E) tests for various Apache Tika d From this directory: ```bash -mvn clean install +./mvnw clean install ``` ## Running All E2E Tests ```bash -mvn test +./mvnw test ``` ## Running Specific Test Module ```bash cd tika-grpc -mvn test +./mvnw test ``` ## Why Standalone? diff --git a/tika-e2e-tests/pom.xml b/tika-e2e-tests/pom.xml index 54f06c403f..4fecac900b 100644 --- a/tika-e2e-tests/pom.xml +++ b/tika-e2e-tests/pom.xml @@ -24,9 +24,14 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd";> <modelVersion>4.0.0</modelVersion> - <groupId>org.apache.tika</groupId> + <parent> + <groupId>org.apache.tika</groupId> + <artifactId>tika-parent</artifactId> + <version>4.0.0-SNAPSHOT</version> + <relativePath>../tika-parent/pom.xml</relativePath> + </parent> + <artifactId>tika-e2e-tests</artifactId> - <version>4.0.0-SNAPSHOT</version> <packaging>pom</packaging> <name>Apache Tika End-to-End Tests</name> <description>End-to-end integration tests for Apache Tika components</description> @@ -42,7 +47,7 @@ <!-- Test dependencies --> <junit.version>5.11.4</junit.version> - <testcontainers.version>1.20.4</testcontainers.version> + <testcontainers.version>2.0.3</testcontainers.version> <!-- Logging --> <slf4j.version>2.0.16</slf4j.version> @@ -82,7 +87,7 @@ </dependency> <dependency> <groupId>org.testcontainers</groupId> - <artifactId>junit-jupiter</artifactId> + <artifactId>testcontainers-junit-jupiter</artifactId> <version>${testcontainers.version}</version> <scope>test</scope> </dependency> @@ -111,6 +116,18 @@ <version>${jackson.version}</version> </dependency> + <!-- Micronaut version alignment for Ignite 3.x --> + <dependency> + <groupId>io.micronaut</groupId> + <artifactId>micronaut-validation</artifactId> + <version>3.10.4</version> + </dependency> + <dependency> + <groupId>io.micronaut</groupId> + <artifactId>micronaut-http</artifactId> + <version>3.10.4</version> + </dependency> + <!-- Lombok --> <dependency> <groupId>org.projectlombok</groupId> @@ -139,5 +156,19 @@ </plugin> </plugins> </pluginManagement> + <plugins> + <plugin> + <groupId>org.apache.rat</groupId> + <artifactId>apache-rat-plugin</artifactId> + <configuration> + <inputExcludes> + <inputExclude>README.md</inputExclude> + <inputExclude>**/README.md</inputExclude> + <inputExclude>**/target/**</inputExclude> + <inputExclude>**/.idea/**</inputExclude> + </inputExcludes> + </configuration> + </plugin> + </plugins> </build> </project> diff --git a/tika-e2e-tests/tika-grpc/README.md b/tika-e2e-tests/tika-grpc/README.md index 12d3fca1b3..63bb173ebc 100644 --- a/tika-e2e-tests/tika-grpc/README.md +++ b/tika-e2e-tests/tika-grpc/README.md @@ -21,7 +21,7 @@ This test module validates the functionality of Apache Tika gRPC Server by: ## Building ```bash -mvn clean install +./mvnw clean install ``` ## Running Tests @@ -29,14 +29,14 @@ mvn clean install ### Run all tests ```bash -mvn test +./mvnw test ``` ### Run specific test ```bash -mvn test -Dtest=FileSystemFetcherTest -mvn test -Dtest=IgniteConfigStoreTest +./mvnw test -Dtest=FileSystemFetcherTest +./mvnw test -Dtest=IgniteConfigStoreTest ``` ### Configure test document range @@ -44,7 +44,7 @@ mvn test -Dtest=IgniteConfigStoreTest By default, only the first batch of GovDocs1 documents (001.zip) is downloaded. To test with more documents: ```bash -mvn test -Dgovdocs1.fromIndex=1 -Dgovdocs1.toIndex=5 +./mvnw test -Dgovdocs1.fromIndex=1 -Dgovdocs1.toIndex=5 ``` This will download and test with batches 001.zip through 005.zip. @@ -54,7 +54,7 @@ This will download and test with batches 001.zip through 005.zip. To limit the test to only process a specific number of documents (useful for quick testing): ```bash -mvn test -Dcorpa.numdocs=10 +./mvnw test -Dcorpa.numdocs=10 ``` This will process only the first 10 documents instead of all documents in the corpus. Omit this parameter or set to -1 to process all documents. @@ -63,13 +63,13 @@ This will process only the first 10 documents instead of all documents in the co ```bash # Test with just 5 documents -mvn test -Dcorpa.numdocs=5 +./mvnw test -Dcorpa.numdocs=5 # Test with 100 documents from multiple batches -mvn test -Dgovdocs1.fromIndex=1 -Dgovdocs1.toIndex=2 -Dcorpa.numdocs=100 +./mvnw test -Dgovdocs1.fromIndex=1 -Dgovdocs1.toIndex=2 -Dcorpa.numdocs=100 # Test all documents (default behavior) -mvn test +./mvnw test ``` ## Test Structure @@ -104,7 +104,7 @@ Or build from the main Tika repository and tag it: ```bash cd /path/to/tika -mvn clean install -DskipTests +./mvnw clean install -DskipTests cd tika-grpc # Follow tika-grpc Docker build instructions ``` diff --git a/tika-e2e-tests/tika-grpc/pom.xml b/tika-e2e-tests/tika-grpc/pom.xml index 7148c37b84..b759dfadde 100644 --- a/tika-e2e-tests/tika-grpc/pom.xml +++ b/tika-e2e-tests/tika-grpc/pom.xml @@ -34,6 +34,13 @@ <name>Apache Tika gRPC End-to-End Tests</name> <description>End-to-end tests for Apache Tika gRPC Server using test containers</description> + <properties> + <!-- Use local server mode by default in CI (faster, no Docker required) --> + <!-- Override with -Dtika.e2e.useLocalServer=false to use Docker --> + <tika.e2e.useLocalServer>true</tika.e2e.useLocalServer> + <corpa.numdocs>2</corpa.numdocs> + </properties> + <dependencies> <!-- Tika gRPC --> <dependency> @@ -87,7 +94,7 @@ </dependency> <dependency> <groupId>org.testcontainers</groupId> - <artifactId>junit-jupiter</artifactId> + <artifactId>testcontainers-junit-jupiter</artifactId> <scope>test</scope> </dependency> @@ -104,6 +111,14 @@ <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> </dependency> + + <!-- Awaitility for robust waiting --> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <version>4.2.0</version> + <scope>test</scope> + </dependency> </dependencies> <build> @@ -122,9 +137,36 @@ <systemPropertyVariables> <govdocs1.fromIndex>1</govdocs1.fromIndex> <govdocs1.toIndex>1</govdocs1.toIndex> + <tika.e2e.useLocalServer>${tika.e2e.useLocalServer}</tika.e2e.useLocalServer> + <corpa.numdocs>${corpa.numdocs}</corpa.numdocs> </systemPropertyVariables> </configuration> </plugin> + <!-- Disable dependency convergence check for e2e tests --> + <!-- Ignite 3.x brings many transitive dependencies that conflict --> + <!-- but tests work fine in practice --> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-enforcer-plugin</artifactId> + <executions> + <execution> + <id>enforce-maven</id> + <phase>none</phase> + </execution> + </executions> + </plugin> + <!-- Configure RAT to exclude files that don't need license headers --> + <plugin> + <groupId>org.apache.rat</groupId> + <artifactId>apache-rat-plugin</artifactId> + <configuration> + <inputExcludes> + <inputExclude>**/README*.md</inputExclude> + <inputExclude>src/test/resources/docker-compose*.yml</inputExclude> + <inputExclude>src/test/resources/log4j2.xml</inputExclude> + </inputExcludes> + </configuration> + </plugin> </plugins> </build> </project> diff --git a/tika-e2e-tests/tika-grpc/sample-configs/ignite/tika-config-ignite.json b/tika-e2e-tests/tika-grpc/sample-configs/ignite/tika-config-ignite.json index 69da03028b..1262f8c549 100644 --- a/tika-e2e-tests/tika-grpc/sample-configs/ignite/tika-config-ignite.json +++ b/tika-e2e-tests/tika-grpc/sample-configs/ignite/tika-config-ignite.json @@ -1,7 +1,7 @@ { "pipes": { "configStoreType": "ignite", - "configStoreParams": "{\n \"cacheName\": \"tika-config-store\",\n \"cacheMode\": \"REPLICATED\",\n \"igniteInstanceName\": \"TikaIgniteCluster\",\n \"autoClose\": true\n }" + "configStoreParams": "{\n \"tableName\": \"tika-config-store\",\n \"igniteInstanceName\": \"TikaIgniteCluster\",\n \"replicas\": 2,\n \"partitions\": 10,\n \"autoClose\": true\n }" }, "fetchers": [ { diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java index 511d671c6c..54a600cc0e 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java @@ -1,20 +1,21 @@ +/* + * 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.tika.pipes; -import com.fasterxml.jackson.databind.ObjectMapper; -import io.grpc.ManagedChannel; -import io.grpc.ManagedChannelBuilder; -import lombok.extern.slf4j.Slf4j; -import org.apache.tika.FetchAndParseReply; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Tag; -import org.junit.jupiter.api.TestInstance; -import org.testcontainers.containers.DockerComposeContainer; -import org.testcontainers.containers.output.Slf4jLogConsumer; -import org.testcontainers.containers.wait.strategy.Wait; -import org.testcontainers.junit.jupiter.Testcontainers; - import java.io.File; import java.io.FileInputStream; import java.io.IOException; @@ -36,9 +37,25 @@ import java.util.stream.Stream; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.TestInstance; +import org.testcontainers.containers.DockerComposeContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Testcontainers; + +import org.apache.tika.FetchAndParseReply; + /** * Base class for Tika gRPC end-to-end tests. - * Uses Docker Compose to start tika-grpc server and runs tests against it. + * Can run with either local server (default in CI) or Docker Compose. */ @TestInstance(TestInstance.Lifecycle.PER_CLASS) @Testcontainers @@ -52,13 +69,137 @@ public abstract class ExternalTestBase { public static final int GOV_DOCS_FROM_IDX = Integer.parseInt(System.getProperty("govdocs1.fromIndex", "1")); public static final int GOV_DOCS_TO_IDX = Integer.parseInt(System.getProperty("govdocs1.toIndex", "1")); public static final String DIGITAL_CORPORA_ZIP_FILES_URL = "https://corp.digitalcorpora.org/corpora/files/govdocs1/zipfiles";; + private static final boolean USE_LOCAL_SERVER = Boolean.parseBoolean(System.getProperty("tika.e2e.useLocalServer", "false")); + private static final int GRPC_PORT = Integer.parseInt(System.getProperty("tika.e2e.grpcPort", "50052")); public static DockerComposeContainer<?> composeContainer; + private static Process localGrpcProcess; @BeforeAll static void setup() throws Exception { loadGovdocs1(); + if (USE_LOCAL_SERVER) { + startLocalGrpcServer(); + } else { + startDockerGrpcServer(); + } + } + + private static void startLocalGrpcServer() throws Exception { + log.info("Starting local tika-grpc server using Maven exec"); + + Path tikaGrpcDir = findTikaGrpcDirectory(); + Path configFile = Path.of("src/test/resources/tika-config.json").toAbsolutePath(); + + if (!Files.exists(configFile)) { + throw new IllegalStateException("Config file not found: " + configFile); + } + + log.info("Using tika-grpc from: {}", tikaGrpcDir); + log.info("Using config file: {}", configFile); + + String javaHome = System.getProperty("java.home"); + boolean isWindows = System.getProperty("os.name").toLowerCase().contains("win"); + String javaCmd = javaHome + (isWindows ? "\\bin\\java.exe" : "/bin/java"); + String mvnCmd = isWindows ? "mvn.cmd" : "mvn"; + + ProcessBuilder pb = new ProcessBuilder( + mvnCmd, + "exec:exec", + "-Dexec.executable=" + javaCmd, + "-Dexec.args=" + + "--add-opens=java.base/java.lang=ALL-UNNAMED " + + "--add-opens=java.base/java.nio=ALL-UNNAMED " + + "--add-opens=java.base/java.util=ALL-UNNAMED " + + "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED " + + "-classpath %classpath " + + "org.apache.tika.pipes.grpc.TikaGrpcServer " + + "-c " + configFile + " " + + "-p " + GRPC_PORT + ); + + pb.directory(tikaGrpcDir.toFile()); + pb.redirectErrorStream(true); + pb.redirectOutput(ProcessBuilder.Redirect.PIPE); + + localGrpcProcess = pb.start(); + + // Start thread to consume output + Thread logThread = new Thread(() -> { + try (java.io.BufferedReader reader = new java.io.BufferedReader( + new java.io.InputStreamReader(localGrpcProcess.getInputStream()))) { + String line; + while ((line = reader.readLine()) != null) { + log.info("tika-grpc: {}", line); + } + } catch (IOException e) { + log.error("Error reading server output", e); + } + }); + logThread.setDaemon(true); + logThread.start(); + + // Wait for server to be ready + waitForServerReady(); + + log.info("Local tika-grpc server started successfully on port {}", GRPC_PORT); + } + + private static Path findTikaGrpcDirectory() { + Path currentDir = Path.of("").toAbsolutePath(); + Path tikaRootDir = currentDir; + + while (tikaRootDir != null && + !(Files.exists(tikaRootDir.resolve("tika-grpc")) && + Files.exists(tikaRootDir.resolve("tika-e2e-tests")))) { + tikaRootDir = tikaRootDir.getParent(); + } + + if (tikaRootDir == null) { + throw new IllegalStateException("Cannot find tika root directory. " + + "Current dir: " + currentDir); + } + + return tikaRootDir.resolve("tika-grpc"); + } + + private static void waitForServerReady() throws Exception { + int maxAttempts = 60; + for (int i = 0; i < maxAttempts; i++) { + try { + ManagedChannel testChannel = ManagedChannelBuilder + .forAddress("localhost", GRPC_PORT) + .usePlaintext() + .build(); + + try { + // Try a simple connection + testChannel.getState(true); + TimeUnit.MILLISECONDS.sleep(100); + if (testChannel.getState(false).toString().contains("READY")) { + log.info("gRPC server is ready!"); + return; + } + } finally { + testChannel.shutdown(); + testChannel.awaitTermination(1, TimeUnit.SECONDS); + } + } catch (Exception e) { + // Server not ready yet + } + TimeUnit.SECONDS.sleep(1); + } + + if (localGrpcProcess != null && localGrpcProcess.isAlive()) { + localGrpcProcess.destroyForcibly(); + } + throw new RuntimeException("Local gRPC server failed to start within timeout"); + } + + private static void startDockerGrpcServer() { + log.info("Starting Docker Compose tika-grpc server"); + composeContainer = new DockerComposeContainer<>( new File("src/test/resources/docker-compose.yml")) .withEnv("HOST_GOVDOCS1_DIR", TEST_FOLDER.getAbsolutePath()) @@ -92,7 +233,18 @@ public abstract class ExternalTestBase { @AfterAll void close() { - if (composeContainer != null) { + if (USE_LOCAL_SERVER && localGrpcProcess != null) { + log.info("Stopping local gRPC server"); + localGrpcProcess.destroy(); + try { + if (!localGrpcProcess.waitFor(10, TimeUnit.SECONDS)) { + localGrpcProcess.destroyForcibly(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + localGrpcProcess.destroyForcibly(); + } + } else if (composeContainer != null) { composeContainer.close(); } } @@ -102,7 +254,7 @@ public abstract class ExternalTestBase { Files.createDirectories(targetDir); for (int i = fromIndex; i <= toIndex; i++) { - String zipName = String.format("%03d.zip", i); + String zipName = String.format(java.util.Locale.ROOT, "%03d.zip", i); String url = DIGITAL_CORPORA_ZIP_FILES_URL + "/" + zipName; Path zipPath = targetDir.resolve(zipName); @@ -172,12 +324,21 @@ public abstract class ExternalTestBase { } public static ManagedChannel getManagedChannel() { - return ManagedChannelBuilder - .forAddress(composeContainer.getServiceHost("tika-grpc", 50052), - composeContainer.getServicePort("tika-grpc", 50052)) - .usePlaintext() - .executor(Executors.newCachedThreadPool()) - .maxInboundMessageSize(160 * 1024 * 1024) - .build(); + if (USE_LOCAL_SERVER) { + return ManagedChannelBuilder + .forAddress("localhost", GRPC_PORT) + .usePlaintext() + .executor(Executors.newCachedThreadPool()) + .maxInboundMessageSize(160 * 1024 * 1024) + .build(); + } else { + return ManagedChannelBuilder + .forAddress(composeContainer.getServiceHost("tika-grpc", 50052), + composeContainer.getServicePort("tika-grpc", 50052)) + .usePlaintext() + .executor(Executors.newCachedThreadPool()) + .maxInboundMessageSize(160 * 1024 * 1024) + .build(); + } } } diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java index d5e6c15ec7..df8f057cc4 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java @@ -1,30 +1,43 @@ +/* + * 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.tika.pipes.filesystem; -import com.fasterxml.jackson.core.JsonProcessingException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + import io.grpc.ManagedChannel; import io.grpc.stub.StreamObserver; import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + import org.apache.tika.FetchAndParseReply; import org.apache.tika.FetchAndParseRequest; -import org.apache.tika.GetFetcherReply; -import org.apache.tika.GetFetcherRequest; import org.apache.tika.SaveFetcherReply; import org.apache.tika.SaveFetcherRequest; import org.apache.tika.TikaGrpc; import org.apache.tika.pipes.ExternalTestBase; import org.apache.tika.pipes.fetcher.fs.FileSystemFetcherConfig; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; - -import java.nio.file.Files; -import java.nio.file.Path; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.UUID; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.stream.Stream; @Slf4j class FileSystemFetcherTest extends ExternalTestBase { @@ -38,10 +51,13 @@ class FileSystemFetcherTest extends ExternalTestBase { // Create and save the fetcher dynamically FileSystemFetcherConfig config = new FileSystemFetcherConfig(); - config.setBasePath("/tika/govdocs1"); + // Use local path when running in local mode, Docker path otherwise + boolean useLocalServer = Boolean.parseBoolean(System.getProperty("tika.e2e.useLocalServer", "false")); + String basePath = useLocalServer ? TEST_FOLDER.getAbsolutePath() : GOV_DOCS_FOLDER; + config.setBasePath(basePath); String configJson = OBJECT_MAPPER.writeValueAsString(config); - log.info("Creating fetcher with config: {}", configJson); + log.info("Creating fetcher with config (basePath={}): {}", basePath, configJson); SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java index f3b9293cb8..d8261b7ecf 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java @@ -1,27 +1,21 @@ +/* + * 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.tika.pipes.ignite; -import com.fasterxml.jackson.databind.ObjectMapper; -import io.grpc.ManagedChannel; -import io.grpc.ManagedChannelBuilder; -import io.grpc.stub.StreamObserver; -import lombok.extern.slf4j.Slf4j; -import org.apache.tika.FetchAndParseReply; -import org.apache.tika.FetchAndParseRequest; -import org.apache.tika.SaveFetcherReply; -import org.apache.tika.SaveFetcherRequest; -import org.apache.tika.TikaGrpc; -import org.apache.tika.pipes.fetcher.fs.FileSystemFetcherConfig; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Tag; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.TestInstance; -import org.testcontainers.containers.DockerComposeContainer; -import org.testcontainers.containers.output.Slf4jLogConsumer; -import org.testcontainers.containers.wait.strategy.Wait; -import org.testcontainers.junit.jupiter.Testcontainers; - import java.io.File; import java.io.FileInputStream; import java.io.IOException; @@ -43,6 +37,31 @@ import java.util.stream.Stream; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.stub.StreamObserver; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.condition.DisabledOnOs; +import org.junit.jupiter.api.condition.OS; +import org.testcontainers.containers.DockerComposeContainer; +import org.testcontainers.containers.output.Slf4jLogConsumer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Testcontainers; + +import org.apache.tika.FetchAndParseReply; +import org.apache.tika.FetchAndParseRequest; +import org.apache.tika.SaveFetcherReply; +import org.apache.tika.SaveFetcherRequest; +import org.apache.tika.TikaGrpc; +import org.apache.tika.pipes.fetcher.fs.FileSystemFetcherConfig; + /** * End-to-end test for Ignite ConfigStore. * Tests that fetchers saved via gRPC are persisted in Ignite @@ -52,6 +71,7 @@ import java.util.zip.ZipInputStream; @Testcontainers @Slf4j @Tag("E2ETest") +@DisabledOnOs(value = OS.WINDOWS, disabledReason = "Maven not on PATH and Docker/Testcontainers not supported on Windows CI") class IgniteConfigStoreTest { private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); @@ -60,16 +80,219 @@ class IgniteConfigStoreTest { private static final int GOV_DOCS_FROM_IDX = Integer.parseInt(System.getProperty("govdocs1.fromIndex", "1")); private static final int GOV_DOCS_TO_IDX = Integer.parseInt(System.getProperty("govdocs1.toIndex", "1")); private static final String DIGITAL_CORPORA_ZIP_FILES_URL = "https://corp.digitalcorpora.org/corpora/files/govdocs1/zipfiles";; + private static final boolean USE_LOCAL_SERVER = Boolean.parseBoolean(System.getProperty("tika.e2e.useLocalServer", "false")); + private static final int GRPC_PORT = Integer.parseInt(System.getProperty("tika.e2e.grpcPort", "50052")); private static DockerComposeContainer<?> igniteComposeContainer; + private static Process localGrpcProcess; @BeforeAll static void setupIgnite() throws Exception { + // Clean up any orphaned processes from previous runs + if (USE_LOCAL_SERVER) { + log.info("Cleaning up any orphaned processes from previous runs"); + try { + killProcessOnPort(GRPC_PORT); + killProcessOnPort(3344); + killProcessOnPort(10800); + } catch (Exception e) { + log.debug("No orphaned processes to clean up"); + } + } + // Load govdocs1 if not already loaded if (!TEST_FOLDER.exists() || TEST_FOLDER.listFiles().length == 0) { downloadAndUnzipGovdocs1(GOV_DOCS_FROM_IDX, GOV_DOCS_TO_IDX); } + if (USE_LOCAL_SERVER) { + startLocalGrpcServer(); + } else { + startDockerGrpcServer(); + } + } + + private static void startLocalGrpcServer() throws Exception { + log.info("Starting local tika-grpc server using Maven"); + + // Find the tika root directory - it should contain both tika-grpc and tika-e2e-tests + Path currentDir = Path.of("").toAbsolutePath(); + Path tikaRootDir = currentDir; + + // Navigate up to find the directory that contains both tika-grpc and tika-e2e-tests + while (tikaRootDir != null && + !(Files.exists(tikaRootDir.resolve("tika-grpc")) && + Files.exists(tikaRootDir.resolve("tika-e2e-tests")))) { + tikaRootDir = tikaRootDir.getParent(); + } + + if (tikaRootDir == null) { + throw new IllegalStateException("Cannot find tika root directory. " + + "Current dir: " + currentDir + ". " + + "Please run from within the tika project."); + } + + Path tikaGrpcDir = tikaRootDir.resolve("tika-grpc"); + if (!Files.exists(tikaGrpcDir)) { + throw new IllegalStateException("Cannot find tika-grpc directory at: " + tikaGrpcDir); + } + + // Use different config for local vs Docker + String configFileName = "tika-config-ignite-local.json"; + Path configFile = Path.of("src/test/resources/" + configFileName).toAbsolutePath(); + + if (!Files.exists(configFile)) { + throw new IllegalStateException("Config file not found: " + configFile); + } + + log.info("Tika root: {}", tikaRootDir); + log.info("Using tika-grpc from: {}", tikaGrpcDir); + log.info("Using config file: {}", configFile); + + // Use mvn exec:exec to run as external process (not exec:java which breaks ServiceLoader) + String javaHome = System.getProperty("java.home"); + String javaCmd = javaHome + "/bin/java"; + + ProcessBuilder pb = new ProcessBuilder( + "mvn", + "exec:exec", + "-Dexec.executable=" + javaCmd, + "-Dexec.args=" + + "--add-opens=java.base/java.lang=ALL-UNNAMED " + + "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED " + + "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED " + + "--add-opens=java.base/java.io=ALL-UNNAMED " + + "--add-opens=java.base/java.nio=ALL-UNNAMED " + + "--add-opens=java.base/java.math=ALL-UNNAMED " + + "--add-opens=java.base/java.util=ALL-UNNAMED " + + "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED " + + "--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED " + + "--add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED " + + "--add-opens=java.base/java.time=ALL-UNNAMED " + + "--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED " + + "--add-opens=java.base/jdk.internal.access=ALL-UNNAMED " + + "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED " + + "--add-opens=java.management/com.sun.jmx.mbeanserver=ALL-UNNAMED " + + "--add-opens=jdk.management/com.sun.management.internal=ALL-UNNAMED " + + "-Dio.netty.tryReflectionSetAccessible=true " + + "-Dignite.work.dir=" + tikaGrpcDir.resolve("target/ignite-work") + " " + + "-classpath %classpath " + + "org.apache.tika.pipes.grpc.TikaGrpcServer " + + "-c " + configFile + " " + + "-p " + GRPC_PORT + ); + + pb.directory(tikaGrpcDir.toFile()); + pb.redirectErrorStream(true); + pb.redirectOutput(ProcessBuilder.Redirect.PIPE); + + localGrpcProcess = pb.start(); + + // Track whether Ignite has started + final boolean[] igniteStarted = {false}; + + // Start a thread to consume and log output, watching for Ignite startup + Thread logThread = new Thread(() -> { + try (java.io.BufferedReader reader = new java.io.BufferedReader( + new java.io.InputStreamReader(localGrpcProcess.getInputStream(), java.nio.charset.StandardCharsets.UTF_8))) { + String line; + while ((line = reader.readLine()) != null) { + log.info("tika-grpc: {}", line); + + // Look for signs that Ignite has fully started + if (line.contains("Ignite server started") || + line.contains("Table") && line.contains("created successfully") || + line.contains("Server started, listening on")) { + synchronized (igniteStarted) { + igniteStarted[0] = true; + igniteStarted.notifyAll(); + } + } + } + } catch (IOException e) { + log.error("Error reading server output", e); + } + }); + logThread.setDaemon(true); + logThread.start(); + + // Wait for Ignite to start - check both log messages and gRPC connectivity + log.info("Waiting for local gRPC server and Ignite to start (timeout: 180 seconds)..."); + + try { + org.awaitility.Awaitility.await() + .atMost(java.time.Duration.ofSeconds(180)) + .pollInterval(java.time.Duration.ofSeconds(2)) + .until(() -> { + boolean igniteReady; + synchronized (igniteStarted) { + igniteReady = igniteStarted[0]; + } + + if (!igniteReady) { + log.debug("Waiting for Ignite to start..."); + return false; + } + + // Try to actually test gRPC readiness with a real (lightweight) call + try { + ManagedChannel testChannel = ManagedChannelBuilder + .forAddress("localhost", GRPC_PORT) + .usePlaintext() + .build(); + + try { + // Try to use the health check service + io.grpc.health.v1.HealthGrpc.HealthBlockingStub healthStub = + io.grpc.health.v1.HealthGrpc.newBlockingStub(testChannel) + .withDeadlineAfter(2, TimeUnit.SECONDS); + + io.grpc.health.v1.HealthCheckResponse response = healthStub.check( + io.grpc.health.v1.HealthCheckRequest.getDefaultInstance()); + + boolean serving = response.getStatus() == + io.grpc.health.v1.HealthCheckResponse.ServingStatus.SERVING; + + if (serving) { + log.info("gRPC server is healthy and serving!"); + return true; + } else { + log.debug("gRPC server responding but not serving yet: {}", response.getStatus()); + return false; + } + } finally { + testChannel.shutdown(); + testChannel.awaitTermination(1, TimeUnit.SECONDS); + } + } catch (io.grpc.StatusRuntimeException e) { + if (e.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) { + // Health check not implemented, just verify channel works + log.info("Health check not available, assuming server is ready"); + return true; + } + log.debug("gRPC server not ready yet: {}", e.getMessage()); + return false; + } catch (Exception e) { + log.debug("gRPC server not ready yet: {}", e.getMessage()); + return false; + } + }); + + log.info("Both gRPC server and Ignite are ready!"); + } catch (org.awaitility.core.ConditionTimeoutException e) { + if (localGrpcProcess.isAlive()) { + localGrpcProcess.destroyForcibly(); + } + throw new RuntimeException("Local gRPC server or Ignite failed to start within timeout", e); + } + + log.info("Local tika-grpc server started successfully on port {}", GRPC_PORT); + } + + + private static void startDockerGrpcServer() { + log.info("Starting Docker Compose tika-grpc server"); + igniteComposeContainer = new DockerComposeContainer<>( new File("src/test/resources/docker-compose-ignite.yml")) .withEnv("HOST_GOVDOCS1_DIR", TEST_FOLDER.getAbsolutePath()) @@ -85,119 +308,221 @@ class IgniteConfigStoreTest { @AfterAll static void teardownIgnite() { - if (igniteComposeContainer != null) { + if (USE_LOCAL_SERVER && localGrpcProcess != null) { + log.info("Stopping local gRPC server and all child processes"); + + try { + // Get the PID of the Maven process + long mvnPid = localGrpcProcess.pid(); + log.info("Maven process PID: {}", mvnPid); + + // Try graceful shutdown first + localGrpcProcess.destroy(); + + if (!localGrpcProcess.waitFor(10, TimeUnit.SECONDS)) { + log.warn("Process didn't stop gracefully, forcing shutdown"); + localGrpcProcess.destroyForcibly(); + localGrpcProcess.waitFor(5, TimeUnit.SECONDS); + } + + // Give it a moment for cleanup + Thread.sleep(2000); + + // Kill any remaining child processes by finding processes listening on Ignite/gRPC ports + // Only do this if the process is still running + try { + killProcessOnPort(GRPC_PORT); // Kill gRPC server + killProcessOnPort(3344); // Kill Ignite's internal port + killProcessOnPort(10800); // Kill Ignite client connector + } catch (Exception e) { + log.debug("Error killing processes on ports (may already be stopped): {}", e.getMessage()); + } + + log.info("Local gRPC server stopped"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + localGrpcProcess.destroyForcibly(); + } + } else if (igniteComposeContainer != null) { igniteComposeContainer.close(); } } + private static void killProcessOnPort(int port) throws IOException, InterruptedException { + // Find process listening on the port using lsof + ProcessBuilder findPb = new ProcessBuilder("lsof", "-ti", ":" + port); + findPb.redirectErrorStream(true); + Process findProcess = findPb.start(); + + try (java.io.BufferedReader reader = new java.io.BufferedReader( + new java.io.InputStreamReader(findProcess.getInputStream(), java.nio.charset.StandardCharsets.UTF_8))) { + String pidStr = reader.readLine(); + if (pidStr != null && !pidStr.trim().isEmpty()) { + long pid = Long.parseLong(pidStr.trim()); + long myPid = ProcessHandle.current().pid(); + + // Don't kill ourselves or our parent + if (pid == myPid || isParentProcess(pid)) { + log.debug("Skipping kill of PID {} on port {} (test process or parent)", pid, port); + return; + } + + log.info("Found process {} listening on port {}, killing it", pid, port); + + ProcessBuilder killPb = new ProcessBuilder("kill", String.valueOf(pid)); + Process killProcess = killPb.start(); + killProcess.waitFor(2, TimeUnit.SECONDS); + + // If still alive, force kill + Thread.sleep(1000); + ProcessBuilder forceKillPb = new ProcessBuilder("kill", "-9", String.valueOf(pid)); + Process forceKillProcess = forceKillPb.start(); + forceKillProcess.waitFor(2, TimeUnit.SECONDS); + } + } + + findProcess.waitFor(2, TimeUnit.SECONDS); + } + + private static boolean isParentProcess(long pid) { + try { + ProcessHandle current = ProcessHandle.current(); + while (current.parent().isPresent()) { + current = current.parent().get(); + if (current.pid() == pid) { + return true; + } + } + } catch (Exception e) { + log.debug("Error checking parent process", e); + } + return false; + } + @Test void testIgniteConfigStore() throws Exception { String fetcherId = "dynamicIgniteFetcher"; ManagedChannel channel = getManagedChannelForIgnite(); - TikaGrpc.TikaBlockingStub blockingStub = TikaGrpc.newBlockingStub(channel); - TikaGrpc.TikaStub tikaStub = TikaGrpc.newStub(channel); - - // Create and save the fetcher dynamically - FileSystemFetcherConfig config = new FileSystemFetcherConfig(); - config.setBasePath("/tika/govdocs1"); - String configJson = OBJECT_MAPPER.writeValueAsString(config); - log.info("Creating fetcher with Ignite ConfigStore: {}", configJson); - - SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest - .newBuilder() - .setFetcherId(fetcherId) - .setFetcherClass("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher") - .setFetcherConfigJson(configJson) - .build()); - - log.info("Fetcher saved to Ignite: {}", saveReply.getFetcherId()); + try { + TikaGrpc.TikaBlockingStub blockingStub = TikaGrpc.newBlockingStub(channel); + TikaGrpc.TikaStub tikaStub = TikaGrpc.newStub(channel); - List<FetchAndParseReply> successes = Collections.synchronizedList(new ArrayList<>()); - List<FetchAndParseReply> errors = Collections.synchronizedList(new ArrayList<>()); + // Create and save the fetcher dynamically + FileSystemFetcherConfig config = new FileSystemFetcherConfig(); + // Use local path when running in local mode, Docker path otherwise + String basePath = USE_LOCAL_SERVER ? TEST_FOLDER.getAbsolutePath() : "/tika/govdocs1"; + config.setBasePath(basePath); + + String configJson = OBJECT_MAPPER.writeValueAsString(config); + log.info("Creating fetcher with Ignite ConfigStore (basePath={}): {}", basePath, configJson); + + SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest + .newBuilder() + .setFetcherId(fetcherId) + .setFetcherClass("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher") + .setFetcherConfigJson(configJson) + .build()); + + log.info("Fetcher saved to Ignite: {}", saveReply.getFetcherId()); + + List<FetchAndParseReply> successes = Collections.synchronizedList(new ArrayList<>()); + List<FetchAndParseReply> errors = Collections.synchronizedList(new ArrayList<>()); + + CountDownLatch countDownLatch = new CountDownLatch(1); + StreamObserver<FetchAndParseRequest> + requestStreamObserver = tikaStub.fetchAndParseBiDirectionalStreaming(new StreamObserver<>() { + @Override + public void onNext(FetchAndParseReply fetchAndParseReply) { + log.debug("Reply from fetch-and-parse - key={}, status={}", + fetchAndParseReply.getFetchKey(), fetchAndParseReply.getStatus()); + if ("FETCH_AND_PARSE_EXCEPTION".equals(fetchAndParseReply.getStatus())) { + errors.add(fetchAndParseReply); + } else { + successes.add(fetchAndParseReply); + } + } - CountDownLatch countDownLatch = new CountDownLatch(1); - StreamObserver<FetchAndParseRequest> - requestStreamObserver = tikaStub.fetchAndParseBiDirectionalStreaming(new StreamObserver<>() { - @Override - public void onNext(FetchAndParseReply fetchAndParseReply) { - log.debug("Reply from fetch-and-parse - key={}, status={}", - fetchAndParseReply.getFetchKey(), fetchAndParseReply.getStatus()); - if ("FETCH_AND_PARSE_EXCEPTION".equals(fetchAndParseReply.getStatus())) { - errors.add(fetchAndParseReply); - } else { - successes.add(fetchAndParseReply); + @Override + public void onError(Throwable throwable) { + log.error("Received an error", throwable); + Assertions.fail(throwable); + countDownLatch.countDown(); } - } - @Override - public void onError(Throwable throwable) { - log.error("Received an error", throwable); - Assertions.fail(throwable); - countDownLatch.countDown(); - } + @Override + public void onCompleted() { + log.info("Finished streaming fetch and parse replies"); + countDownLatch.countDown(); + } + }); - @Override - public void onCompleted() { - log.info("Finished streaming fetch and parse replies"); - countDownLatch.countDown(); + // Submit files for parsing - limit to configured number + int maxDocs = Integer.parseInt(System.getProperty("corpa.numdocs", "-1")); + log.info("Document limit: {}", maxDocs == -1 ? "unlimited" : maxDocs); + + try (Stream<Path> paths = Files.walk(TEST_FOLDER.toPath())) { + Stream<Path> fileStream = paths.filter(Files::isRegularFile); + + if (maxDocs > 0) { + fileStream = fileStream.limit(maxDocs); + } + + fileStream.forEach(file -> { + try { + String relPath = TEST_FOLDER.toPath().relativize(file).toString(); + requestStreamObserver.onNext(FetchAndParseRequest + .newBuilder() + .setFetcherId(fetcherId) + .setFetchKey(relPath) + .build()); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); } - }); + log.info("Done submitting files to Ignite-backed fetcher {}", fetcherId); - // Submit files for parsing - limit to configured number - int maxDocs = Integer.parseInt(System.getProperty("corpa.numdocs", "-1")); - log.info("Document limit: {}", maxDocs == -1 ? "unlimited" : maxDocs); - - try (Stream<Path> paths = Files.walk(TEST_FOLDER.toPath())) { - Stream<Path> fileStream = paths.filter(Files::isRegularFile); + requestStreamObserver.onCompleted(); + + // Wait for all parsing to complete + try { + if (!countDownLatch.await(3, TimeUnit.MINUTES)) { + log.error("Timed out waiting for parse to complete"); + Assertions.fail("Timed out waiting for parsing to complete"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + Assertions.fail("Interrupted while waiting for parsing to complete"); + } - if (maxDocs > 0) { - fileStream = fileStream.limit(maxDocs); + // Verify documents were processed + if (maxDocs == -1) { + assertAllFilesFetched(TEST_FOLDER.toPath(), successes, errors); + } else { + int totalProcessed = successes.size() + errors.size(); + log.info("Processed {} documents with Ignite ConfigStore (limit was {})", + totalProcessed, maxDocs); + Assertions.assertTrue(totalProcessed <= maxDocs, + "Should not process more than " + maxDocs + " documents"); + Assertions.assertTrue(totalProcessed > 0, + "Should have processed at least one document"); } - fileStream.forEach(file -> { - try { - String relPath = TEST_FOLDER.toPath().relativize(file).toString(); - requestStreamObserver.onNext(FetchAndParseRequest - .newBuilder() - .setFetcherId(fetcherId) - .setFetchKey(relPath) - .build()); - } catch (Exception e) { - throw new RuntimeException(e); + log.info("Ignite ConfigStore test completed successfully - {} successes, {} errors", + successes.size(), errors.size()); + } finally { + // Properly shutdown gRPC channel to avoid resource leak + channel.shutdown(); + try { + if (!channel.awaitTermination(5, TimeUnit.SECONDS)) { + channel.shutdownNow(); } - }); - } - log.info("Done submitting files to Ignite-backed fetcher {}", fetcherId); - - requestStreamObserver.onCompleted(); - - // Wait for all parsing to complete - try { - if (!countDownLatch.await(3, TimeUnit.MINUTES)) { - log.error("Timed out waiting for parse to complete"); - Assertions.fail("Timed out waiting for parsing to complete"); + } catch (InterruptedException e) { + channel.shutdownNow(); + Thread.currentThread().interrupt(); } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - Assertions.fail("Interrupted while waiting for parsing to complete"); } - - // Verify documents were processed - if (maxDocs == -1) { - assertAllFilesFetched(TEST_FOLDER.toPath(), successes, errors); - } else { - int totalProcessed = successes.size() + errors.size(); - log.info("Processed {} documents with Ignite ConfigStore (limit was {})", - totalProcessed, maxDocs); - Assertions.assertTrue(totalProcessed <= maxDocs, - "Should not process more than " + maxDocs + " documents"); - Assertions.assertTrue(totalProcessed > 0, - "Should have processed at least one document"); - } - - log.info("Ignite ConfigStore test completed successfully - {} successes, {} errors", - successes.size(), errors.size()); } // Helper method for downloading test data @@ -206,7 +531,7 @@ class IgniteConfigStoreTest { Files.createDirectories(targetDir); for (int i = fromIndex; i <= toIndex; i++) { - String zipName = String.format("%03d.zip", i); + String zipName = String.format(java.util.Locale.ROOT, "%03d.zip", i); String url = DIGITAL_CORPORA_ZIP_FILES_URL + "/" + zipName; Path zipPath = targetDir.resolve(zipName); @@ -277,12 +602,21 @@ class IgniteConfigStoreTest { // Helper method to create gRPC channel private static ManagedChannel getManagedChannelForIgnite() { - return ManagedChannelBuilder - .forAddress(igniteComposeContainer.getServiceHost("tika-grpc", 50052), - igniteComposeContainer.getServicePort("tika-grpc", 50052)) - .usePlaintext() - .executor(Executors.newCachedThreadPool()) - .maxInboundMessageSize(160 * 1024 * 1024) - .build(); + if (USE_LOCAL_SERVER) { + return ManagedChannelBuilder + .forAddress("localhost", GRPC_PORT) + .usePlaintext() + .executor(Executors.newCachedThreadPool()) + .maxInboundMessageSize(160 * 1024 * 1024) + .build(); + } else { + return ManagedChannelBuilder + .forAddress(igniteComposeContainer.getServiceHost("tika-grpc", 50052), + igniteComposeContainer.getServicePort("tika-grpc", 50052)) + .usePlaintext() + .executor(Executors.newCachedThreadPool()) + .maxInboundMessageSize(160 * 1024 * 1024) + .build(); + } } } diff --git a/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite.json b/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite-local.json similarity index 87% copy from tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite.json copy to tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite-local.json index 2cca83ceaf..18c407346b 100644 --- a/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite.json +++ b/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite-local.json @@ -3,7 +3,7 @@ "pipes": { "numClients": 1, "configStoreType": "ignite", - "configStoreParams": "{\"cacheName\": \"tika-e2e-test\", \"cacheMode\": \"REPLICATED\", \"igniteInstanceName\": \"TikaE2ETest\", \"autoClose\": true}", + "configStoreParams": "{\"tableName\": \"tika_e2e_test\", \"igniteInstanceName\": \"TikaE2ETest\", \"replicas\": 2, \"partitions\": 10, \"autoClose\": true}", "forkedJvmArgs": [ "--add-opens=java.base/jdk.internal.access=ALL-UNNAMED", "--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED", @@ -35,7 +35,7 @@ { "fs": { "staticFetcher": { - "basePath": "/tika/govdocs1" + "basePath": "/home/ndipiazza/source/github/apache/tika/tika_e2e_tests/tika-grpc/target/govdocs1" } } } diff --git a/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite.json b/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite.json index 2cca83ceaf..e39b9fb2a2 100644 --- a/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite.json +++ b/tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite.json @@ -3,7 +3,7 @@ "pipes": { "numClients": 1, "configStoreType": "ignite", - "configStoreParams": "{\"cacheName\": \"tika-e2e-test\", \"cacheMode\": \"REPLICATED\", \"igniteInstanceName\": \"TikaE2ETest\", \"autoClose\": true}", + "configStoreParams": "{\"tableName\": \"tika_e2e_test\", \"igniteInstanceName\": \"TikaE2ETest\", \"replicas\": 2, \"partitions\": 10, \"autoClose\": true}", "forkedJvmArgs": [ "--add-opens=java.base/jdk.internal.access=ALL-UNNAMED", "--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED",
