Copilot commented on code in PR #2961: URL: https://github.com/apache/tika/pull/2961#discussion_r3916770701
########## tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/DocumentBuilder.java: ########## @@ -0,0 +1,168 @@ +/* + * 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 lossless tagged tail for everything else. + */ +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 without a typed home + // yet: the reply's fields map still carries the flat content, and the structured + // content tree is a planned additive follow-up -- duplicating the whole body + // into `extra` as a string would defeat both. + 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}) 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" -> + ParseStatus.Status.PARTIAL; + default -> ParseStatus.Status.FAILED; + }; Review Comment: mapPipesStatus() does not handle PipesResult status "PARTIAL_TIMEOUT" (a success-category status in PipesResult.RESULT_STATUS); it currently falls through to FAILED, which misclassifies truncated-but-successful parses. ########## tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java: ########## @@ -357,65 +380,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; } Review Comment: On InterruptedException, runFetchAndParse() returns null; callers then complete the RPC without sending any reply message (fetchAndParseImpl() just returns), which is invalid for unary RPCs and can confuse streaming clients. Prefer returning an explicit failure outcome (status/error) so v1/v2 always send a terminal reply. ########## tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcV2ServerImpl.java: ########## @@ -0,0 +1,124 @@ +/* + * 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.grpc; + +import io.grpc.stub.StreamObserver; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.tika.grpc.mapper.DocumentBuilder; +import org.apache.tika.grpc.v2.FetchAndParseReply; +import org.apache.tika.grpc.v2.FetchAndParseRequest; +import org.apache.tika.grpc.v2.TikaV2Grpc; +import org.apache.tika.parser.ParseContext; + +/** + * Experimental v2 parse surface. Reuses the same pipes/fetcher runtime as the v1 + * {@link TikaGrpcServerImpl}; fetcher management stays on v1. Replies carry the typed + * {@link org.apache.tika.grpc.v2.Document} contract instead of the legacy fields map. + */ +class TikaGrpcV2ServerImpl extends TikaV2Grpc.TikaV2ImplBase { + + private static final Logger LOG = LoggerFactory.getLogger(TikaGrpcV2ServerImpl.class); + + private final TikaGrpcServerImpl v1; + + TikaGrpcV2ServerImpl(TikaGrpcServerImpl v1) { + this.v1 = v1; + } + + @Override + public void fetchAndParse(FetchAndParseRequest request, + StreamObserver<FetchAndParseReply> responseObserver) { + if (v1.denyPerRequestConfig(request.getAdditionalFetchConfigJson(), + request.getParseContextJson(), responseObserver)) { + return; + } + ParseContext parseContext = v1.buildRequestParseContext(request.getFetcherId(), + request.getAdditionalFetchConfigJson(), request.getParseContextJson(), + responseObserver); + if (parseContext == null) { + return; + } + fetchAndParseImpl(request, parseContext, responseObserver); + responseObserver.onCompleted(); + } + + @Override + public void fetchAndParseServerSideStreaming(FetchAndParseRequest request, + StreamObserver<FetchAndParseReply> responseObserver) { + if (v1.denyPerRequestConfig(request.getAdditionalFetchConfigJson(), + request.getParseContextJson(), responseObserver)) { + return; + } + ParseContext parseContext = v1.buildRequestParseContext(request.getFetcherId(), + request.getAdditionalFetchConfigJson(), request.getParseContextJson(), + responseObserver); + if (parseContext == null) { + return; + } + fetchAndParseImpl(request, parseContext, responseObserver); + } Review Comment: fetchAndParseServerSideStreaming() writes a response via onNext but never calls responseObserver.onCompleted(), so server-side streaming clients can hang waiting for stream completion. -- 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]
