Skip to content

Commit e8c36c9

Browse files
nddipiazzaCopilot
nddipiazza
and
Copilot
authored
TIKA-4722: Add parse_context_json field to FetchAndParseRequest for per-request ParseContext configuration (#2797)
- Replace handler_type field with generic parse_context_json string field (field 5) - parse_context_json accepts a JSON object mapping parse-context component names to their configs Example: {"basic-content-handler-factory": {"type": "HTML"}} - TikaGrpcServerImpl iterates the JSON fields and calls parseContext.setJsonConfig() for each - Allows overriding any registered parse-context component per request, not just content handler type - e2e test uses parse_context_json to request HTML and TEXT output and verifies the difference Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent 0300d94 commit e8c36c9

4 files changed

Lines changed: 401 additions & 0 deletions

File tree

‎tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java‎

Lines changed: 356 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,356 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.tika.pipes.filesystem;
18+
19+
import java.io.BufferedReader;
20+
import java.io.File;
21+
import java.io.IOException;
22+
import java.io.InputStreamReader;
23+
import java.nio.charset.StandardCharsets;
24+
import java.nio.file.Files;
25+
import java.nio.file.Path;
26+
import java.util.Locale;
27+
import java.util.concurrent.TimeUnit;
28+
29+
import io.grpc.ManagedChannel;
30+
import io.grpc.ManagedChannelBuilder;
31+
import lombok.extern.slf4j.Slf4j;
32+
import org.awaitility.Awaitility;
33+
import org.junit.jupiter.api.AfterAll;
34+
import org.junit.jupiter.api.Assertions;
35+
import org.junit.jupiter.api.BeforeAll;
36+
import org.junit.jupiter.api.Tag;
37+
import org.junit.jupiter.api.Test;
38+
import org.junit.jupiter.api.TestInstance;
39+
import org.junit.jupiter.api.condition.DisabledOnOs;
40+
import org.junit.jupiter.api.condition.OS;
41+
42+
import org.apache.tika.FetchAndParseReply;
43+
import org.apache.tika.FetchAndParseRequest;
44+
import org.apache.tika.SaveFetcherReply;
45+
import org.apache.tika.SaveFetcherRequest;
46+
import org.apache.tika.TikaGrpc;
47+
import org.apache.tika.pipes.ExternalTestBase;
48+
import org.apache.tika.pipes.fetcher.fs.FileSystemFetcherConfig;
49+
50+
/**
51+
* Tests per-request ParseContext configuration via FetchAndParseRequest.parse_context_json.
52+
*
53+
* Uses the Ignite ConfigStore so that fetchers registered via saveFetcher are visible
54+
* to both the gRPC server JVM and the forked PipesServer JVM.
55+
*
56+
* Verifies that clients can override any parse context component on a per-request basis
57+
* by providing a JSON object with component names as keys.
58+
* Example: {"basic-content-handler-factory": {"type": "HTML"}}
59+
*/
60+
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
61+
@Slf4j
62+
@Tag("E2ETest")
63+
@DisabledOnOs(value = OS.WINDOWS, disabledReason = "exec:exec classpath exceeds Windows CreateProcess command-line length limit")
64+
class HandlerTypeTest {
65+
66+
private static final File TEST_FOLDER = ExternalTestBase.TEST_FOLDER;
67+
private static final int GRPC_PORT = Integer.parseInt(System.getProperty("tika.e2e.grpcPort", "50052"));
68+
69+
private static Process localGrpcProcess;
70+
71+
@BeforeAll
72+
void setup() throws Exception {
73+
try {
74+
killProcessOnPort(GRPC_PORT);
75+
killProcessOnPort(3344);
76+
killProcessOnPort(10800);
77+
} catch (Exception e) {
78+
log.debug("No orphaned processes to clean up: {}", e.getMessage());
79+
}
80+
81+
ExternalTestBase.copyTestFixtures();
82+
startLocalGrpcServer();
83+
}
84+
85+
@AfterAll
86+
void teardown() {
87+
if (localGrpcProcess != null) {
88+
log.info("Stopping local gRPC server and child processes");
89+
localGrpcProcess.destroy();
90+
try {
91+
if (!localGrpcProcess.waitFor(10, TimeUnit.SECONDS)) {
92+
localGrpcProcess.destroyForcibly();
93+
localGrpcProcess.waitFor(5, TimeUnit.SECONDS);
94+
}
95+
Thread.sleep(2000);
96+
killProcessOnPort(GRPC_PORT);
97+
killProcessOnPort(3344);
98+
killProcessOnPort(10800);
99+
} catch (Exception e) {
100+
log.debug("Error during teardown: {}", e.getMessage());
101+
}
102+
log.info("Local gRPC server stopped");
103+
}
104+
}
105+
106+
private static void startLocalGrpcServer() throws Exception {
107+
log.info("Starting local tika-grpc server with Ignite config for HandlerType test");
108+
109+
Path currentDir = Path.of("").toAbsolutePath();
110+
Path tikaRootDir = currentDir;
111+
while (tikaRootDir != null &&
112+
!(Files.exists(tikaRootDir.resolve("tika-grpc")) &&
113+
Files.exists(tikaRootDir.resolve("tika-e2e-tests")))) {
114+
tikaRootDir = tikaRootDir.getParent();
115+
}
116+
if (tikaRootDir == null) {
117+
throw new IllegalStateException("Cannot find tika root directory. Current dir: " + currentDir);
118+
}
119+
120+
Path tikaGrpcDir = tikaRootDir.resolve("tika-grpc");
121+
Path configFile = Path.of("src/test/resources/tika-config-ignite-handlertype.json").toAbsolutePath();
122+
if (!Files.exists(configFile)) {
123+
throw new IllegalStateException("Config file not found: " + configFile);
124+
}
125+
126+
log.info("tika-grpc dir: {}", tikaGrpcDir);
127+
log.info("Config file: {}", configFile);
128+
129+
String javaHome = System.getProperty("java.home");
130+
boolean isWindows = System.getProperty("os.name").toLowerCase(Locale.ROOT).contains("win");
131+
String javaCmd = javaHome + (isWindows ? "\\bin\\java.exe" : "/bin/java");
132+
String mvnCmd = tikaRootDir.resolve(isWindows ? "mvnw.cmd" : "mvnw").toString();
133+
134+
ProcessBuilder pb = new ProcessBuilder(
135+
mvnCmd,
136+
"exec:exec",
137+
"-Dexec.executable=" + javaCmd,
138+
"-Dexec.args=" +
139+
"--add-opens=java.base/java.lang=ALL-UNNAMED " +
140+
"--add-opens=java.base/java.lang.invoke=ALL-UNNAMED " +
141+
"--add-opens=java.base/java.lang.reflect=ALL-UNNAMED " +
142+
"--add-opens=java.base/java.io=ALL-UNNAMED " +
143+
"--add-opens=java.base/java.nio=ALL-UNNAMED " +
144+
"--add-opens=java.base/java.math=ALL-UNNAMED " +
145+
"--add-opens=java.base/java.util=ALL-UNNAMED " +
146+
"--add-opens=java.base/java.util.concurrent=ALL-UNNAMED " +
147+
"--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED " +
148+
"--add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED " +
149+
"--add-opens=java.base/java.time=ALL-UNNAMED " +
150+
"--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED " +
151+
"--add-opens=java.base/jdk.internal.access=ALL-UNNAMED " +
152+
"--add-opens=java.base/sun.nio.ch=ALL-UNNAMED " +
153+
"--add-opens=java.management/com.sun.jmx.mbeanserver=ALL-UNNAMED " +
154+
"--add-opens=jdk.management/com.sun.management.internal=ALL-UNNAMED " +
155+
"-Dio.netty.tryReflectionSetAccessible=true " +
156+
"-Dignite.work.dir=\"" + tikaGrpcDir.resolve("target/ignite-work-handlertype") + "\" " +
157+
"-classpath %classpath " +
158+
"org.apache.tika.pipes.grpc.TikaGrpcServer " +
159+
"-c \"" + configFile + "\" " +
160+
"-p " + GRPC_PORT
161+
);
162+
163+
pb.directory(tikaGrpcDir.toFile());
164+
pb.redirectErrorStream(true);
165+
pb.redirectOutput(ProcessBuilder.Redirect.PIPE);
166+
167+
localGrpcProcess = pb.start();
168+
169+
final boolean[] igniteStarted = {false};
170+
Thread logThread = new Thread(() -> {
171+
try (BufferedReader reader = new BufferedReader(
172+
new InputStreamReader(localGrpcProcess.getInputStream(), StandardCharsets.UTF_8))) {
173+
String line;
174+
while ((line = reader.readLine()) != null) {
175+
log.info("tika-grpc: {}", line);
176+
if (line.contains("Ignite server started") ||
177+
(line.contains("Table") && line.contains("created successfully")) ||
178+
line.contains("Server started, listening on")) {
179+
synchronized (igniteStarted) {
180+
igniteStarted[0] = true;
181+
igniteStarted.notifyAll();
182+
}
183+
}
184+
}
185+
} catch (IOException e) {
186+
log.error("Error reading server output", e);
187+
}
188+
});
189+
logThread.setDaemon(true);
190+
logThread.start();
191+
192+
try {
193+
Awaitility.await()
194+
.atMost(java.time.Duration.ofSeconds(180))
195+
.pollInterval(java.time.Duration.ofSeconds(2))
196+
.until(() -> {
197+
synchronized (igniteStarted) {
198+
if (!igniteStarted[0]) {
199+
return false;
200+
}
201+
}
202+
ManagedChannel testChannel = ManagedChannelBuilder
203+
.forAddress("localhost", GRPC_PORT)
204+
.usePlaintext()
205+
.build();
206+
try {
207+
io.grpc.health.v1.HealthGrpc.HealthBlockingStub healthStub =
208+
io.grpc.health.v1.HealthGrpc.newBlockingStub(testChannel)
209+
.withDeadlineAfter(2, TimeUnit.SECONDS);
210+
io.grpc.health.v1.HealthCheckResponse response = healthStub.check(
211+
io.grpc.health.v1.HealthCheckRequest.getDefaultInstance());
212+
return response.getStatus() ==
213+
io.grpc.health.v1.HealthCheckResponse.ServingStatus.SERVING;
214+
} catch (io.grpc.StatusRuntimeException e) {
215+
if (e.getStatus().getCode() == io.grpc.Status.Code.UNIMPLEMENTED) {
216+
return true;
217+
}
218+
return false;
219+
} catch (Exception e) {
220+
return false;
221+
} finally {
222+
testChannel.shutdown();
223+
testChannel.awaitTermination(1, TimeUnit.SECONDS);
224+
}
225+
});
226+
} catch (org.awaitility.core.ConditionTimeoutException e) {
227+
if (localGrpcProcess.isAlive()) {
228+
localGrpcProcess.destroyForcibly();
229+
}
230+
throw new RuntimeException("tika-grpc server with Ignite failed to start within timeout", e);
231+
}
232+
233+
log.info("HandlerType test server ready on port {}", GRPC_PORT);
234+
}
235+
236+
private ManagedChannel getManagedChannel() {
237+
return ManagedChannelBuilder
238+
.forAddress("localhost", GRPC_PORT)
239+
.usePlaintext()
240+
.maxInboundMessageSize(160 * 1024 * 1024)
241+
.build();
242+
}
243+
244+
private static void killProcessOnPort(int port) throws IOException, InterruptedException {
245+
ProcessBuilder findPb = new ProcessBuilder("lsof", "-ti", ":" + port);
246+
findPb.redirectErrorStream(true);
247+
Process findProcess = findPb.start();
248+
try (BufferedReader reader = new BufferedReader(
249+
new InputStreamReader(findProcess.getInputStream(), StandardCharsets.UTF_8))) {
250+
String pidStr = reader.readLine();
251+
if (pidStr != null && !pidStr.trim().isEmpty()) {
252+
long pid = Long.parseLong(pidStr.trim());
253+
long myPid = ProcessHandle.current().pid();
254+
if (pid == myPid || isParentProcess(pid)) {
255+
return;
256+
}
257+
String cmdLine = ProcessHandle.of(pid)
258+
.flatMap(h -> h.info().commandLine())
259+
.orElse("");
260+
if (!cmdLine.contains("tika") && !cmdLine.contains("TikaGrpc") && !cmdLine.contains("ignite")) {
261+
log.debug("Skipping kill of PID {} on port {} — not a tika/ignite process", pid, port);
262+
return;
263+
}
264+
log.info("Killing tika/ignite process {} on port {}", pid, port);
265+
new ProcessBuilder("kill", String.valueOf(pid)).start().waitFor(2, TimeUnit.SECONDS);
266+
Thread.sleep(1000);
267+
new ProcessBuilder("kill", "-9", String.valueOf(pid)).start().waitFor(2, TimeUnit.SECONDS);
268+
}
269+
}
270+
findProcess.waitFor(2, TimeUnit.SECONDS);
271+
}
272+
273+
private static boolean isParentProcess(long pid) {
274+
try {
275+
ProcessHandle current = ProcessHandle.current();
276+
while (current.parent().isPresent()) {
277+
current = current.parent().get();
278+
if (current.pid() == pid) {
279+
return true;
280+
}
281+
}
282+
} catch (Exception e) {
283+
log.debug("Error checking parent process", e);
284+
}
285+
return false;
286+
}
287+
288+
@Test
289+
void testParseContextJson() throws Exception {
290+
String fetcherId = "handlerTypeFetcher";
291+
ManagedChannel channel = getManagedChannel();
292+
try {
293+
TikaGrpc.TikaBlockingStub blockingStub = TikaGrpc.newBlockingStub(channel);
294+
295+
FileSystemFetcherConfig config = new FileSystemFetcherConfig();
296+
config.setBasePath(TEST_FOLDER.getAbsolutePath());
297+
298+
SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest.newBuilder()
299+
.setFetcherId(fetcherId)
300+
.setFetcherClass("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher")
301+
.setFetcherConfigJson(ExternalTestBase.OBJECT_MAPPER.writeValueAsString(config))
302+
.build());
303+
log.info("Fetcher created: {}", saveReply.getFetcherId());
304+
305+
// Parse sample.html requesting HTML output
306+
FetchAndParseReply htmlReply = blockingStub.fetchAndParse(FetchAndParseRequest.newBuilder()
307+
.setFetcherId(fetcherId)
308+
.setFetchKey("sample.html")
309+
.setParseContextJson("{\"basic-content-handler-factory\": {\"type\": \"HTML\"}}")
310+
.build());
311+
312+
log.info("HTML parse status: {}", htmlReply.getStatus());
313+
Assertions.assertEquals("PARSE_SUCCESS", htmlReply.getStatus(),
314+
"Parse should succeed with HTML handler type");
315+
316+
String htmlContent = htmlReply.getFieldsMap().get("X-TIKA:content");
317+
Assertions.assertNotNull(htmlContent, "Content should be present in HTML response");
318+
log.info("HTML content (first 200 chars): {}", htmlContent.substring(0, Math.min(200, htmlContent.length())));
319+
Assertions.assertTrue(
320+
htmlContent.contains("<html") || htmlContent.contains("<body") || htmlContent.contains("<p"),
321+
"HTML handler should produce HTML markup, got: " + htmlContent);
322+
323+
// Parse the same file requesting plain text — expect no HTML tags
324+
FetchAndParseReply textReply = blockingStub.fetchAndParse(FetchAndParseRequest.newBuilder()
325+
.setFetcherId(fetcherId)
326+
.setFetchKey("sample.html")
327+
.setParseContextJson("{\"basic-content-handler-factory\": {\"type\": \"TEXT\"}}")
328+
.build());
329+
330+
log.info("Text parse status: {}", textReply.getStatus());
331+
Assertions.assertEquals("PARSE_SUCCESS", textReply.getStatus(),
332+
"Parse should succeed with TEXT handler type");
333+
334+
String textContent = textReply.getFieldsMap().get("X-TIKA:content");
335+
Assertions.assertNotNull(textContent, "Content should be present in text response");
336+
log.info("Text content (first 200 chars): {}", textContent.substring(0, Math.min(200, textContent.length())));
337+
Assertions.assertFalse(
338+
textContent.contains("<html") || textContent.contains("<body"),
339+
"TEXT handler should not produce HTML tags, got: " + textContent);
340+
341+
Assertions.assertNotEquals(htmlContent, textContent,
342+
"HTML and TEXT outputs should differ for the same document");
343+
344+
} finally {
345+
channel.shutdown();
346+
try {
347+
if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
348+
channel.shutdownNow();
349+
}
350+
} catch (InterruptedException e) {
351+
channel.shutdownNow();
352+
Thread.currentThread().interrupt();
353+
}
354+
}
355+
}
356+
}

‎tika-e2e-tests/tika-grpc/src/test/resources/tika-config-ignite-handlertype.json‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
{
2+
"plugin-roots": ["/var/cache/tika/plugins"],
3+
"pipes": {
4+
"numClients": 1,
5+
"configStoreType": "ignite",
6+
"configStoreParams": "{\"tableName\": \"tika_handlertype_test\", \"igniteInstanceName\": \"TikaHandlerTypeTest\", \"replicas\": 1, \"partitions\": 10, \"autoClose\": true}",
7+
"forkedJvmArgs": [
8+
"--add-opens=java.base/jdk.internal.access=ALL-UNNAMED",
9+
"--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED",
10+
"--add-opens=java.base/sun.nio.ch=ALL-UNNAMED",
11+
"--add-opens=java.base/sun.util.calendar=ALL-UNNAMED",
12+
"--add-opens=java.management/com.sun.jmx.mbeanserver=ALL-UNNAMED",
13+
"--add-opens=jdk.internal.jvmstat/sun.jvmstat.monitor=ALL-UNNAMED",
14+
"--add-opens=java.base/sun.reflect.generics.reflectiveObjects=ALL-UNNAMED",
15+
"--add-opens=jdk.management/com.sun.management.internal=ALL-UNNAMED",
16+
"--add-opens=java.base/java.io=ALL-UNNAMED",
17+
"--add-opens=java.base/java.nio=ALL-UNNAMED",
18+
"--add-opens=java.base/java.net=ALL-UNNAMED",
19+
"--add-opens=java.base/java.util=ALL-UNNAMED",
20+
"--add-opens=java.base/java.util.concurrent=ALL-UNNAMED",
21+
"--add-opens=java.base/java.util.concurrent.locks=ALL-UNNAMED",
22+
"--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED",
23+
"--add-opens=java.base/java.lang=ALL-UNNAMED",
24+
"--add-opens=java.base/java.lang.invoke=ALL-UNNAMED",
25+
"--add-opens=java.base/java.math=ALL-UNNAMED",
26+
"--add-opens=java.sql/java.sql=ALL-UNNAMED",
27+
"--add-opens=java.base/java.lang.reflect=ALL-UNNAMED",
28+
"--add-opens=java.base/java.time=ALL-UNNAMED",
29+
"--add-opens=java.base/java.text=ALL-UNNAMED",
30+
"--add-opens=java.management/sun.management=ALL-UNNAMED",
31+
"--add-opens=java.desktop/java.awt.font=ALL-UNNAMED"
32+
]
33+
}
34+
}

0 commit comments

Comments
 (0)