Copilot commented on code in PR #2961: URL: https://github.com/apache/tika/pull/2961#discussion_r4355599346
########## tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/transform/MetadataTagger.java: ########## @@ -0,0 +1,232 @@ +/* + * 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.grpc.mapper.transform; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.Locale; +import java.util.Set; + +import com.google.protobuf.Timestamp; + +import org.apache.tika.grpc.v2.BooleanValues; +import org.apache.tika.grpc.v2.Document; +import org.apache.tika.grpc.v2.IntegerValues; +import org.apache.tika.grpc.v2.MetadataField; +import org.apache.tika.grpc.v2.MetadataValue; +import org.apache.tika.grpc.v2.NumberValues; +import org.apache.tika.grpc.v2.StringValues; +import org.apache.tika.grpc.v2.TimestampValues; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.metadata.Property; +import org.apache.tika.serialization.MetadataValueTypes; +import org.apache.tika.utils.DateUtils; + +/** + * Builds the tagged metadata tail. A Tika metadata entry is an array of values under + * one key ({@link Metadata} is String[]-backed), so every tagged value is a typed + * ARRAY: the tag comes from Tika's <em>declared</em> {@link Property} value type and + * applies to every element -- a declared integer sequence like {@code pdf:charsPerPage} + * arrives as int64s, one per page, not as strings. Keys with no declared type are + * emitted as strings, never guessed -- so we never turn "8 8 8 8" into a broken + * integer. If any element refuses its declared type, the whole entry falls back to + * strings: lossless beats typed when the two conflict. + */ +public final class MetadataTagger { + + /** + * {@link Property#get(String)} can only see Property constants whose declaring class + * has already been initialized by the JVM. In the gRPC server the parse runs in a + * forked pipes process, so nothing in the server JVM necessarily touches e.g. + * {@code org.apache.tika.metadata.PDF} before mapping -- and declared types + * ({@code pdf:encrypted} is a boolean, ...) would silently degrade to plain strings + * depending on class-loading order. Force-initialize the tika-core property + * vocabularies once so tagging is deterministic. + */ + private static final Class<?>[] PROPERTY_VOCABULARIES = { + org.apache.tika.metadata.AccessPermissions.class, + org.apache.tika.metadata.ClimateForecast.class, + org.apache.tika.metadata.Database.class, + org.apache.tika.metadata.DublinCore.class, + org.apache.tika.metadata.DWG.class, + org.apache.tika.metadata.Epub.class, + org.apache.tika.metadata.ExternalProcess.class, + org.apache.tika.metadata.FileSystem.class, + org.apache.tika.metadata.Font.class, + org.apache.tika.metadata.Geographic.class, + org.apache.tika.metadata.HTML.class, + org.apache.tika.metadata.HttpHeaders.class, + org.apache.tika.metadata.IPTC.class, + org.apache.tika.metadata.MachineMetadata.class, + org.apache.tika.metadata.MAPI.class, + org.apache.tika.metadata.Message.class, + org.apache.tika.metadata.Office.class, + org.apache.tika.metadata.OfficeOpenXMLCore.class, + org.apache.tika.metadata.OfficeOpenXMLExtended.class, + org.apache.tika.metadata.PageAnchoring.class, + org.apache.tika.metadata.PagedText.class, + org.apache.tika.metadata.PDF.class, + org.apache.tika.metadata.Photoshop.class, + org.apache.tika.metadata.PST.class, + org.apache.tika.metadata.QuattroPro.class, + org.apache.tika.metadata.Rendering.class, + org.apache.tika.metadata.RTFMetadata.class, + org.apache.tika.metadata.TIFF.class, + org.apache.tika.metadata.TikaCoreProperties.class, + org.apache.tika.metadata.TikaPagedText.class, + org.apache.tika.metadata.WARC.class, + org.apache.tika.metadata.WordPerfect.class, + org.apache.tika.metadata.XMP.class, + org.apache.tika.metadata.XMPDC.class, + org.apache.tika.metadata.XMPDM.class, + org.apache.tika.metadata.XMPIdq.class, + org.apache.tika.metadata.XMPMM.class, + org.apache.tika.metadata.XMPPDF.class, + org.apache.tika.metadata.XMPRights.class, + org.apache.tika.metadata.Zip.class, + }; + + static { + for (Class<?> vocabulary : PROPERTY_VOCABULARIES) { + try { + // A class literal alone does NOT initialize the class; forName(…, true, …) does. + Class.forName(vocabulary.getName(), true, vocabulary.getClassLoader()); + } catch (ClassNotFoundException e) { + throw new ExceptionInInitializerError(e); + } + } + } + + // The same date formats Metadata.getDate uses; DateUtils is not thread-safe, so + // access is synchronized exactly as tika-core does (see TIKA-495). + private static final DateUtils DATE_UTILS = new DateUtils(); + + private MetadataTagger() { + } + + /** Append every key not already consumed by a typed field into {@code Document.extra}. */ + public static void appendTail(Metadata tika, Set<String> consumed, Document.Builder document) { + for (String key : tika.names()) { + if (consumed.contains(key)) { + continue; + } + MetadataValue value = tag(tika, key); + if (value != null) { + document.addExtra(MetadataField.newBuilder().setKey(key).setValue(value).build()); + } + } + } + + /** + * Tag one key's value array by its declared Property element type, or by the type a pipes + * fork reported for a key declared only there; fall back to strings for keys of unknown + * type or elements the type cannot parse. + */ + public static MetadataValue tag(Metadata tika, String key) { + String[] raw = tika.getValues(key); + if (raw == null || raw.length == 0) { + return null; + } + List<String> values = new ArrayList<>(raw.length); + for (String v : raw) { + if (v != null && !v.trim().isEmpty()) { + values.add(v.trim()); + } + } + if (values.isEmpty()) { + return null; + } + Property property = Property.get(key); + // a composite's own type is PROPERTY; tag by its primary's type + Property.ValueType type = property != null + ? property.getPrimaryProperty().getValueType() + : MetadataValueTypes.get(key); + if (type != null) { + MetadataValue typed = tagByDeclaredType(type, values); + if (typed != null) { + return typed; + } + } + StringValues.Builder strings = StringValues.newBuilder(); + strings.addAllValues(values); + return MetadataValue.newBuilder().setStrings(strings).build(); + } + + private static MetadataValue tagByDeclaredType(Property.ValueType type, List<String> values) { + switch (type) { + case INTEGER: { + IntegerValues.Builder integers = IntegerValues.newBuilder(); + for (String v : values) { + try { + integers.addValues(Long.parseLong(v)); + } catch (NumberFormatException ignored) { + return null; + } + } + return MetadataValue.newBuilder().setIntegers(integers).build(); + } + case REAL: + case RATIONAL: { + NumberValues.Builder numbers = NumberValues.newBuilder(); + for (String v : values) { + try { + numbers.addValues(Double.parseDouble(v)); + } catch (NumberFormatException ignored) { + return null; + } + } + return MetadataValue.newBuilder().setNumbers(numbers).build(); + } + case BOOLEAN: { + BooleanValues.Builder booleans = BooleanValues.newBuilder(); + for (String v : values) { + booleans.addValues(parseBoolean(v)); + } + return MetadataValue.newBuilder().setBooleans(booleans).build(); + } Review Comment: `parseBoolean` never fails, so any unrecognized boolean token (e.g., \"maybe\") will be silently coerced to `false` instead of triggering the documented fallback-to-strings behavior when a value can't be parsed as its declared type. Make `parseBoolean` validate inputs (e.g., return `Boolean`/`Optional<Boolean>` and return null/empty for unknown tokens) and have `tagByDeclaredType(BOOLEAN, …)` return `null` on unknown values so the whole entry correctly falls back to strings. ########## tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/transform/MetadataTagger.java: ########## @@ -0,0 +1,232 @@ +/* + * 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.grpc.mapper.transform; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.Locale; +import java.util.Set; + +import com.google.protobuf.Timestamp; + +import org.apache.tika.grpc.v2.BooleanValues; +import org.apache.tika.grpc.v2.Document; +import org.apache.tika.grpc.v2.IntegerValues; +import org.apache.tika.grpc.v2.MetadataField; +import org.apache.tika.grpc.v2.MetadataValue; +import org.apache.tika.grpc.v2.NumberValues; +import org.apache.tika.grpc.v2.StringValues; +import org.apache.tika.grpc.v2.TimestampValues; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.metadata.Property; +import org.apache.tika.serialization.MetadataValueTypes; +import org.apache.tika.utils.DateUtils; + +/** + * Builds the tagged metadata tail. A Tika metadata entry is an array of values under + * one key ({@link Metadata} is String[]-backed), so every tagged value is a typed + * ARRAY: the tag comes from Tika's <em>declared</em> {@link Property} value type and + * applies to every element -- a declared integer sequence like {@code pdf:charsPerPage} + * arrives as int64s, one per page, not as strings. Keys with no declared type are + * emitted as strings, never guessed -- so we never turn "8 8 8 8" into a broken + * integer. If any element refuses its declared type, the whole entry falls back to + * strings: lossless beats typed when the two conflict. + */ +public final class MetadataTagger { + + /** + * {@link Property#get(String)} can only see Property constants whose declaring class + * has already been initialized by the JVM. In the gRPC server the parse runs in a + * forked pipes process, so nothing in the server JVM necessarily touches e.g. + * {@code org.apache.tika.metadata.PDF} before mapping -- and declared types + * ({@code pdf:encrypted} is a boolean, ...) would silently degrade to plain strings + * depending on class-loading order. Force-initialize the tika-core property + * vocabularies once so tagging is deterministic. + */ + private static final Class<?>[] PROPERTY_VOCABULARIES = { + org.apache.tika.metadata.AccessPermissions.class, + org.apache.tika.metadata.ClimateForecast.class, + org.apache.tika.metadata.Database.class, + org.apache.tika.metadata.DublinCore.class, + org.apache.tika.metadata.DWG.class, + org.apache.tika.metadata.Epub.class, + org.apache.tika.metadata.ExternalProcess.class, + org.apache.tika.metadata.FileSystem.class, + org.apache.tika.metadata.Font.class, + org.apache.tika.metadata.Geographic.class, + org.apache.tika.metadata.HTML.class, + org.apache.tika.metadata.HttpHeaders.class, + org.apache.tika.metadata.IPTC.class, + org.apache.tika.metadata.MachineMetadata.class, + org.apache.tika.metadata.MAPI.class, + org.apache.tika.metadata.Message.class, + org.apache.tika.metadata.Office.class, + org.apache.tika.metadata.OfficeOpenXMLCore.class, + org.apache.tika.metadata.OfficeOpenXMLExtended.class, + org.apache.tika.metadata.PageAnchoring.class, + org.apache.tika.metadata.PagedText.class, + org.apache.tika.metadata.PDF.class, + org.apache.tika.metadata.Photoshop.class, + org.apache.tika.metadata.PST.class, + org.apache.tika.metadata.QuattroPro.class, + org.apache.tika.metadata.Rendering.class, + org.apache.tika.metadata.RTFMetadata.class, + org.apache.tika.metadata.TIFF.class, + org.apache.tika.metadata.TikaCoreProperties.class, + org.apache.tika.metadata.TikaPagedText.class, + org.apache.tika.metadata.WARC.class, + org.apache.tika.metadata.WordPerfect.class, + org.apache.tika.metadata.XMP.class, + org.apache.tika.metadata.XMPDC.class, + org.apache.tika.metadata.XMPDM.class, + org.apache.tika.metadata.XMPIdq.class, + org.apache.tika.metadata.XMPMM.class, + org.apache.tika.metadata.XMPPDF.class, + org.apache.tika.metadata.XMPRights.class, + org.apache.tika.metadata.Zip.class, + }; + + static { + for (Class<?> vocabulary : PROPERTY_VOCABULARIES) { + try { + // A class literal alone does NOT initialize the class; forName(…, true, …) does. + Class.forName(vocabulary.getName(), true, vocabulary.getClassLoader()); + } catch (ClassNotFoundException e) { + throw new ExceptionInInitializerError(e); + } + } + } + + // The same date formats Metadata.getDate uses; DateUtils is not thread-safe, so + // access is synchronized exactly as tika-core does (see TIKA-495). + private static final DateUtils DATE_UTILS = new DateUtils(); + + private MetadataTagger() { + } + + /** Append every key not already consumed by a typed field into {@code Document.extra}. */ + public static void appendTail(Metadata tika, Set<String> consumed, Document.Builder document) { + for (String key : tika.names()) { + if (consumed.contains(key)) { + continue; + } + MetadataValue value = tag(tika, key); + if (value != null) { + document.addExtra(MetadataField.newBuilder().setKey(key).setValue(value).build()); + } + } + } + + /** + * Tag one key's value array by its declared Property element type, or by the type a pipes + * fork reported for a key declared only there; fall back to strings for keys of unknown + * type or elements the type cannot parse. + */ + public static MetadataValue tag(Metadata tika, String key) { + String[] raw = tika.getValues(key); + if (raw == null || raw.length == 0) { + return null; + } + List<String> values = new ArrayList<>(raw.length); + for (String v : raw) { + if (v != null && !v.trim().isEmpty()) { + values.add(v.trim()); + } + } + if (values.isEmpty()) { + return null; + } + Property property = Property.get(key); + // a composite's own type is PROPERTY; tag by its primary's type + Property.ValueType type = property != null + ? property.getPrimaryProperty().getValueType() + : MetadataValueTypes.get(key); + if (type != null) { + MetadataValue typed = tagByDeclaredType(type, values); + if (typed != null) { + return typed; + } + } + StringValues.Builder strings = StringValues.newBuilder(); + strings.addAllValues(values); + return MetadataValue.newBuilder().setStrings(strings).build(); + } + + private static MetadataValue tagByDeclaredType(Property.ValueType type, List<String> values) { + switch (type) { + case INTEGER: { + IntegerValues.Builder integers = IntegerValues.newBuilder(); + for (String v : values) { + try { + integers.addValues(Long.parseLong(v)); + } catch (NumberFormatException ignored) { + return null; + } + } + return MetadataValue.newBuilder().setIntegers(integers).build(); + } + case REAL: + case RATIONAL: { + NumberValues.Builder numbers = NumberValues.newBuilder(); + for (String v : values) { + try { + numbers.addValues(Double.parseDouble(v)); + } catch (NumberFormatException ignored) { + return null; + } + } + return MetadataValue.newBuilder().setNumbers(numbers).build(); + } + case BOOLEAN: { + BooleanValues.Builder booleans = BooleanValues.newBuilder(); + for (String v : values) { + booleans.addValues(parseBoolean(v)); + } + return MetadataValue.newBuilder().setBooleans(booleans).build(); + } + case DATE: { + TimestampValues.Builder timestamps = TimestampValues.newBuilder(); + for (String v : values) { + Date date = parseDate(v); + if (date == null) { + return null; + } + timestamps.addValues(toTimestamp(date)); + } + return MetadataValue.newBuilder().setTimestamps(timestamps).build(); + } + default: + return null; // not a declared scalar type -> caller falls back to strings + } + } + + private static synchronized Date parseDate(String value) { + return DATE_UTILS.tryToParse(value); + } + + static boolean parseBoolean(String v) { + String s = v.trim().toLowerCase(Locale.ROOT); + return "true".equals(s) || "yes".equals(s) || "1".equals(s); + } Review Comment: `parseBoolean` never fails, so any unrecognized boolean token (e.g., \"maybe\") will be silently coerced to `false` instead of triggering the documented fallback-to-strings behavior when a value can't be parsed as its declared type. Make `parseBoolean` validate inputs (e.g., return `Boolean`/`Optional<Boolean>` and return null/empty for unknown tokens) and have `tagByDeclaredType(BOOLEAN, …)` return `null` on unknown values so the whole entry correctly falls back to strings. ########## tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java: ########## @@ -360,65 +383,120 @@ private ParseContext buildRequestParseContext(FetchAndParseRequest request, return null; } } - String additionalFetchConfigJson = request.getAdditionalFetchConfigJson(); if (StringUtils.isNotBlank(additionalFetchConfigJson)) { // The fork reads this jsonConfig by the fetcher's registered component name // (e.g. "http-fetcher"). An unregistered key fails the fork's resolveAll and a // wire-blocked one is refused by its restricted tuple deserialization -- both // exit the worker. Enforce here: a 400, not a JVM restart per request. - var info = ComponentNameResolver.getComponentInfo(request.getFetcherId()); + var info = ComponentNameResolver.getComponentInfo(fetcherId); if (info.isEmpty() || ComponentNameResolver.isWireBlocked( ComponentNameResolver.determineContextKey(info.get()))) { responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT .withDescription("additional_fetch_config_json requires fetcher_id to " + "be a registered, wire-allowed component name; got '" - + request.getFetcherId() + "'") + + fetcherId + "'") .asRuntimeException()); return null; } - parseContext.setJsonConfig(request.getFetcherId(), additionalFetchConfigJson); + parseContext.setJsonConfig(fetcherId, additionalFetchConfigJson); } return parseContext; } private void fetchAndParseImpl(FetchAndParseRequest request, ParseContext parseContext, StreamObserver<FetchAndParseReply> responseObserver) { + FetchParseOutcome outcome = runFetchAndParse( + request.getFetcherId(), request.getFetchKey(), parseContext); + if (outcome == null) { + return; + } + FetchAndParseReply.Builder fetchReplyBuilder = + FetchAndParseReply.newBuilder() + .setFetchKey(outcome.fetchKey()) + .setStatus(outcome.status()) + .putAllFields(outcome.fields()); + if (outcome.errorMessage() != null) { + fetchReplyBuilder.setErrorMessage(outcome.errorMessage()); + } + responseObserver.onNext(fetchReplyBuilder.build()); + } + + /** + * Shared pipes round-trip used by the v1 {@code fields}-map reply and the v2 typed + * {@code Document} reply. {@code parseContext} is the per-request context already + * validated by {@link #buildRequestParseContext}. Returns primary metadata as + * {@code null} when the pipes result carried no metadata list (so the v2 builder can + * distinguish empty output from an empty {@link Metadata} object). + */ + FetchParseOutcome runFetchAndParse(String fetcherId, String fetchKey, + ParseContext parseContext) { Fetcher fetcher; try { - fetcher = fetcherManager.getFetcher(request.getFetcherId()); + fetcher = fetcherManager.getFetcher(fetcherId); } catch (TikaException | IOException e) { - throw new RuntimeException("Could not find fetcher with name " + request.getFetcherId(), e); + throw new RuntimeException("Could not find fetcher with name " + fetcherId, e); } Metadata tikaMetadata = new Metadata(); + // Times the whole pipesParser.parse() round trip: fetch and parse both happen + // inside the forked pipes worker, so this is fetch+parse latency, not parse-only. + long fetchParseStart = System.nanoTime(); try { - PipesResult pipesResult = pipesParser.parse(new FetchEmitTuple(request.getFetchKey(), new FetchKey(fetcher.getExtensionConfig().id(), request.getFetchKey()), - new EmitKey(), tikaMetadata, parseContext, FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP)); - FetchAndParseReply.Builder fetchReplyBuilder = - FetchAndParseReply.newBuilder() - .setFetchKey(request.getFetchKey()) - .setStatus(pipesResult.status().name()); - if (pipesResult.status().equals(PipesResult.RESULT_STATUS.FETCH_EXCEPTION)) { - fetchReplyBuilder.setErrorMessage(pipesResult.message()); - } + PipesResult pipesResult = pipesParser.parse(new FetchEmitTuple( + fetchKey, + new FetchKey(fetcher.getExtensionConfig().id(), fetchKey), + new EmitKey(), + tikaMetadata, + parseContext, + FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP)); + long fetchParseTimeMs = (System.nanoTime() - fetchParseStart) / 1_000_000L; + Map<String, String> fields = new LinkedHashMap<>(); + Metadata primary = null; if (pipesResult.emitData() != null && pipesResult.emitData().getMetadataList() != null) { for (Metadata metadata : pipesResult.emitData().getMetadataList()) { for (String name : metadata.names()) { String value = metadata.get(name); if (value != null) { - fetchReplyBuilder.putFields(name, value); + fields.put(name, value); } } } + if (!pipesResult.emitData().getMetadataList().isEmpty()) { + primary = pipesResult.emitData().getMetadataList().get(0); + } } - responseObserver.onNext(fetchReplyBuilder.build()); + String errorMessage = null; + if (pipesResult.status().equals(PipesResult.RESULT_STATUS.FETCH_EXCEPTION)) { + errorMessage = pipesResult.message(); + } + return new FetchParseOutcome( + fetchKey, + pipesResult.status().name(), + errorMessage, + fields, + primary, + fetchParseTimeMs); } catch (IOException | PipesException e) { throw new RuntimeException(e); } catch (InterruptedException e) { Thread.currentThread().interrupt(); + return null; } } + /** + * Outcome of a single fetch+parse pipes round trip, shared by v1 and v2 reply builders. + * {@code null} means the calling thread was interrupted before a reply could be built. Review Comment: On `InterruptedException`, `runFetchAndParse` returns `null`, and `fetchAndParseImpl` then returns without calling `onNext`/`onError`. For unary RPCs this can result in the method returning and the outer handler completing the call without sending a response message (violating unary semantics) or leaving the client with an unexpected error. Prefer propagating an explicit gRPC cancellation/error (e.g., `Status.CANCELLED`) or returning an outcome that results in a single reply with a failure status, and ensure the outer RPC handler does not call `onCompleted()` after an error/cancellation path. ########## tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java: ########## @@ -360,65 +383,120 @@ private ParseContext buildRequestParseContext(FetchAndParseRequest request, return null; } } - String additionalFetchConfigJson = request.getAdditionalFetchConfigJson(); if (StringUtils.isNotBlank(additionalFetchConfigJson)) { // The fork reads this jsonConfig by the fetcher's registered component name // (e.g. "http-fetcher"). An unregistered key fails the fork's resolveAll and a // wire-blocked one is refused by its restricted tuple deserialization -- both // exit the worker. Enforce here: a 400, not a JVM restart per request. - var info = ComponentNameResolver.getComponentInfo(request.getFetcherId()); + var info = ComponentNameResolver.getComponentInfo(fetcherId); if (info.isEmpty() || ComponentNameResolver.isWireBlocked( ComponentNameResolver.determineContextKey(info.get()))) { responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT .withDescription("additional_fetch_config_json requires fetcher_id to " + "be a registered, wire-allowed component name; got '" - + request.getFetcherId() + "'") + + fetcherId + "'") .asRuntimeException()); return null; } - parseContext.setJsonConfig(request.getFetcherId(), additionalFetchConfigJson); + parseContext.setJsonConfig(fetcherId, additionalFetchConfigJson); } return parseContext; } private void fetchAndParseImpl(FetchAndParseRequest request, ParseContext parseContext, StreamObserver<FetchAndParseReply> responseObserver) { + FetchParseOutcome outcome = runFetchAndParse( + request.getFetcherId(), request.getFetchKey(), parseContext); + if (outcome == null) { Review Comment: On `InterruptedException`, `runFetchAndParse` returns `null`, and `fetchAndParseImpl` then returns without calling `onNext`/`onError`. For unary RPCs this can result in the method returning and the outer handler completing the call without sending a response message (violating unary semantics) or leaving the client with an unexpected error. Prefer propagating an explicit gRPC cancellation/error (e.g., `Status.CANCELLED`) or returning an outcome that results in a single reply with a failure status, and ensure the outer RPC handler does not call `onCompleted()` after an error/cancellation path. ########## tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java: ########## @@ -638,6 +771,78 @@ public void onCompleted() { } } + /** + * The v2 server-streaming variant must close the call like the v1 one above: one typed + * reply, no error, exactly one completion. Same in-process transport and direct + * executor, so the counts are final when the stub call returns. + */ + @Test + public void testV2ServerSideStreamingSendsTerminalSignal(Resources resources) throws Exception { + String serverName = InProcessServerBuilder.generateName(); + TikaGrpcServerImpl serviceImpl = newService(tikaConfigUnlocked); + Server server = InProcessServerBuilder + .forName(serverName) + .directExecutor() + .addService(serviceImpl) + .addService(new TikaGrpcV2ServerImpl(serviceImpl)) + .build() + .start(); + resources.register(server, Duration.ofSeconds(10)); + + ManagedChannel channel = InProcessChannelBuilder + .forName(serverName) + .directExecutor() + .build(); + resources.register(channel, Duration.ofSeconds(10)); + TikaV2Grpc.TikaV2Stub v2 = TikaV2Grpc.newStub(channel); + + String fetcherId = createFetcherId(1); + String fetchKey = "tika4766-stream-" + UUID.randomUUID() + ".html"; + File testFile = new File("target", fetchKey); + FileUtils.writeStringToFile(testFile, + "<html><body>terminal signal</body></html>", StandardCharsets.UTF_8); + + List<org.apache.tika.grpc.v2.FetchAndParseReply> replies = + Collections.synchronizedList(new ArrayList<>()); + AtomicInteger errors = new AtomicInteger(); + AtomicInteger completions = new AtomicInteger(); + StreamObserver<org.apache.tika.grpc.v2.FetchAndParseReply> observer = + new StreamObserver<>() { + @Override + public void onNext(org.apache.tika.grpc.v2.FetchAndParseReply reply) { + replies.add(reply); + } + + @Override + public void onError(Throwable throwable) { + errors.incrementAndGet(); + } + + @Override + public void onCompleted() { + completions.incrementAndGet(); + } + }; + + try { + v2.fetchAndParseServerSideStreaming(org.apache.tika.grpc.v2.FetchAndParseRequest + .newBuilder() + .setFetcherId(fetcherId) + .setFetchKey(fetchKey) + .build(), observer); + + assertEquals(1, replies.size(), "one reply for one fetch key"); + assertEquals(PipesResult.RESULT_STATUS.PARSE_SUCCESS.name(), + replies.get(0).getDocument().getStatus().getPipesStatus(), + "the fixture must actually parse, or this test proves nothing"); + assertEquals(0, errors.get(), "no error on the happy path"); + assertEquals(1, completions.get(), + "server streaming must send a terminal signal"); Review Comment: This asserts reply/error/completion counts immediately after invoking the async streaming stub. Even with `directExecutor()`, delivery timing can still be sensitive and lead to flaky tests. Use a synchronization primitive (e.g., `CountDownLatch` completed in `onCompleted`/`onError`) and await with a timeout before asserting `replies.size()`, `errors`, and `completions`. ########## tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/DocumentBuilder.java: ########## @@ -0,0 +1,169 @@ +/* + * 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.grpc.mapper; + +import java.time.Instant; +import java.util.HashSet; +import java.util.Set; + +import org.apache.tika.Tika; +import org.apache.tika.digest.DigestDef; +import org.apache.tika.grpc.mapper.transform.DocumentTransformers; +import org.apache.tika.grpc.v2.Document; +import org.apache.tika.grpc.v2.ParseStatus; +import org.apache.tika.grpc.v2.SourceOrigin; +import org.apache.tika.metadata.HttpHeaders; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.metadata.TikaCoreProperties; + +/** + * Builds a {@link Document} from Tika's parse output: the envelope (content type, + * origin, status), the typed Dublin Core metadata via {@link DocumentTransformers}, + * and the tagged tail for the remaining metadata. + */ +public final class DocumentBuilder { + + private static final DocumentTransformers TRANSFORMERS = DocumentTransformers.defaults(); + + // The key the parse pipeline records the source digest under when a digester is + // configured. Derived from DigestDef so it cannot drift from the producer: a + // hand-rolled "...SHA256" copy of it did, silently, when the algorithm keys moved + // to their Java names ("SHA-256"). + private static final String SHA256_DIGEST_KEY = + new DigestDef(DigestDef.Algorithm.SHA256, DigestDef.Encoding.HEX).metadataKey(); + + private DocumentBuilder() { + } + + /** + * Maps Tika parse output to {@link Document}. Pass {@code primary == null} when the + * pipes result carried no metadata at all (e.g. a fetch failure, or a crash before + * the parse produced anything): the returned Document then has only an id and a + * status. The status still derives from {@code pipesStatus} -- {@code EMPTY_OUTPUT} + * with no metadata is a legitimate success -- and an explicit error is recorded for + * the non-success cases. + */ + public static Document build(Metadata primary, String docId, String pipesStatus, + long fetchParseTimeMs) { + ParseStatus.Builder status = ParseStatus.newBuilder() + .setStatus(mapPipesStatus(pipesStatus)) + .setFetchParseTimeMs(fetchParseTimeMs) + .setTikaVersion(Tika.getString()); + if (pipesStatus != null && !pipesStatus.isEmpty()) { + status.setPipesStatus(pipesStatus); + } + + Document.Builder document = Document.newBuilder(); + if (docId != null && !docId.isEmpty()) { + document.setId(docId); + } + + if (primary == null) { + if (status.getStatus() != ParseStatus.Status.SUCCESS) { + status.addErrors("No metadata returned from parse"); + } + return document.setStatus(status.build()).build(); + } + + // Keys the envelope maps below are consumed up front so the tagged tail never + // carries them a second time. tk:content is consumed too: the Document does not + // carry the extracted text, and copying the whole body into `extra` as a string + // would bloat every reply. + Set<String> consumed = new HashSet<>(); + consumed.add(TikaCoreProperties.TIKA_CONTENT.getName()); + + String contentType = primary.get(HttpHeaders.CONTENT_TYPE); + if (contentType != null && !contentType.isBlank()) { + document.setContentType(contentType.trim()); + consumed.add(HttpHeaders.CONTENT_TYPE.getName()); + } + + Instant now = Instant.now(); + document.setParsedAt(com.google.protobuf.Timestamp.newBuilder() + .setSeconds(now.getEpochSecond()) + .setNanos(now.getNano()) + .build()); + + SourceOrigin.Builder origin = SourceOrigin.newBuilder(); + String resourceName = primary.get(TikaCoreProperties.RESOURCE_NAME_KEY); + if (resourceName != null && !resourceName.isBlank()) { + origin.setFilename(resourceName.trim()); + consumed.add(TikaCoreProperties.RESOURCE_NAME_KEY.getName()); + } + String contentLength = primary.get(HttpHeaders.CONTENT_LENGTH); + if (contentLength != null && !contentLength.isBlank()) { + try { + origin.setByteSize(Long.parseLong(contentLength.trim())); + consumed.add(HttpHeaders.CONTENT_LENGTH.getName()); + } catch (NumberFormatException ignored) { + // leave unconsumed; falls through to the tagged tail + } + } + String parserClass = primary.get(TikaCoreProperties.TIKA_PARSED_BY); + if (parserClass != null && !parserClass.isBlank()) { + origin.setParser(parserClass.trim()); + } + String sha256 = primary.get(SHA256_DIGEST_KEY); + if (sha256 != null && !sha256.isBlank()) { + origin.setSha256(sha256.trim()); + consumed.add(SHA256_DIGEST_KEY); + } + document.setOrigin(origin.build()); + + String[] parsedByFull = primary.getValues(TikaCoreProperties.TIKA_PARSED_BY_FULL_SET); + if (parsedByFull == null || parsedByFull.length == 0) { + parsedByFull = primary.getValues(TikaCoreProperties.TIKA_PARSED_BY); + } + for (String parser : parsedByFull) { + if (parser != null && !parser.isBlank()) { + status.addParsersUsed(parser.trim()); + } + } + consumed.add(TikaCoreProperties.TIKA_PARSED_BY.getName()); + consumed.add(TikaCoreProperties.TIKA_PARSED_BY_FULL_SET.getName()); + document.setStatus(status.build()); + + TRANSFORMERS.transform(primary, document, consumed); + + return document.build(); + } + + /** + * Maps {@code org.apache.tika.pipes.api.PipesResult.RESULT_STATUS.name()} to + * {@link ParseStatus.Status}. Values are named after the real enum constants (not + * guessed) since this module intentionally has no compile dependency on tika-pipes-api: + * a clean success (e.g. {@code PARSE_SUCCESS}, {@code EMIT_SUCCESS}) is {@code SUCCESS}; + * a success that happened alongside a caught exception (e.g. + * {@code PARSE_SUCCESS_WITH_EXCEPTION}) or that the deadline cut short but still emitted + * ({@code PARTIAL_TIMEOUT}) is {@code PARTIAL}; everything else -- including process + * crashes like {@code TIMEOUT} and {@code OOM}, which are not partial successes -- is + * {@code FAILED}. + */ + private static ParseStatus.Status mapPipesStatus(String pipesStatus) { + if (pipesStatus == null || pipesStatus.isEmpty()) { + return ParseStatus.Status.UNSPECIFIED; + } + return switch (pipesStatus) { + case "EMPTY_OUTPUT", "PARSE_SUCCESS", "EMIT_SUCCESS", "EMIT_SUCCESS_PASSBACK" -> + ParseStatus.Status.SUCCESS; + case "PARSE_SUCCESS_WITH_EXCEPTION", "PARSE_EXCEPTION_NO_EMIT", "EMIT_SUCCESS_PARSE_EXCEPTION", + "PARTIAL_TIMEOUT" -> Review Comment: The method Javadoc says `PARTIAL` is for success-category statuses with caveats / partial emission, while everything else is `FAILED`. Mapping `PARSE_EXCEPTION_NO_EMIT` to `PARTIAL` appears inconsistent with that description (and the name strongly implies no output was emitted). Consider mapping `PARSE_EXCEPTION_NO_EMIT` to `FAILED` (or update the Javadoc + add a focused unit test) so `ParseStatus.Status` remains aligned with the documented categorization. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
