Copilot commented on code in PR #2961: URL: https://github.com/apache/tika/pull/2961#discussion_r4362918963
########## tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/DocumentBuilder.java: ########## @@ -0,0 +1,171 @@ +/* + * 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.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()); + } + } Review Comment: `primary.getValues(...)` can return null for both `TIKA_PARSED_BY_FULL_SET` and `TIKA_PARSED_BY`. In that case, iterating `for (String parser : parsedByFull)` will throw an NPE for documents that have metadata but no parsed-by keys. Guard the loop (e.g., treat null as an empty array) before iterating. ########## tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcV2ServerImpl.java: ########## @@ -0,0 +1,148 @@ +/* + * 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 java.util.function.BiFunction; + +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.Document; +import org.apache.tika.grpc.v2.FetchAndParseBiDirectionalStreamingRequest; +import org.apache.tika.grpc.v2.FetchAndParseBiDirectionalStreamingResponse; +import org.apache.tika.grpc.v2.FetchAndParseRequest; +import org.apache.tika.grpc.v2.FetchAndParseResponse; +import org.apache.tika.grpc.v2.FetchAndParseServerSideStreamingRequest; +import org.apache.tika.grpc.v2.FetchAndParseServerSideStreamingResponse; +import org.apache.tika.grpc.v2.TikaV2ServiceGrpc; +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 Document} contract instead of the v1 fields map. + */ +class TikaGrpcV2ServerImpl extends TikaV2ServiceGrpc.TikaV2ServiceImplBase { + + 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<FetchAndParseResponse> responseObserver) { + FetchRequest fetch = new FetchRequest(request.getFetcherId(), request.getFetchKey(), + request.getAdditionalFetchConfigJson(), request.getParseContextJson()); + if (reply(fetch, responseObserver, (fetchKey, document) -> FetchAndParseResponse + .newBuilder() + .setFetchKey(fetchKey) + .setDocument(document) + .build())) { + responseObserver.onCompleted(); + } + } + + @Override + public void fetchAndParseServerSideStreaming(FetchAndParseServerSideStreamingRequest request, + StreamObserver<FetchAndParseServerSideStreamingResponse> responseObserver) { + FetchRequest fetch = new FetchRequest(request.getFetcherId(), request.getFetchKey(), + request.getAdditionalFetchConfigJson(), request.getParseContextJson()); + if (reply(fetch, responseObserver, (fetchKey, document) -> + FetchAndParseServerSideStreamingResponse + .newBuilder() + .setFetchKey(fetchKey) + .setDocument(document) + .build())) { + responseObserver.onCompleted(); + } + } + + @Override + public StreamObserver<FetchAndParseBiDirectionalStreamingRequest> fetchAndParseBiDirectionalStreaming( + StreamObserver<FetchAndParseBiDirectionalStreamingResponse> responseObserver) { + return new StreamObserver<>() { + @Override + public void onNext(FetchAndParseBiDirectionalStreamingRequest request) { + FetchRequest fetch = new FetchRequest(request.getFetcherId(), + request.getFetchKey(), request.getAdditionalFetchConfigJson(), + request.getParseContextJson()); + reply(fetch, responseObserver, (fetchKey, document) -> + FetchAndParseBiDirectionalStreamingResponse + .newBuilder() + .setFetchKey(fetchKey) + .setDocument(document) + .build()); + } + + @Override + public void onError(Throwable throwable) { + LOG.error("Parse error occurred", throwable); + } + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } Review Comment: In the bidirectional streaming handler, `onError` only logs and does not terminate the response stream. This can leave clients waiting indefinitely for completion. Propagate the error to the client (e.g., call `responseObserver.onError(throwable)`) and consider preventing further `onNext` processing after an error. ########## tika-grpc-mapper/src/main/java/org/apache/tika/grpc/mapper/transform/MetadataTagger.java: ########## @@ -0,0 +1,243 @@ +/* + * 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); + } + } + } Review Comment: This static initializer forces initialization of a large set of tika-core vocabulary classes at class-load time. That imposes a guaranteed startup/first-use cost and makes classloading side effects unconditional. Consider lazy initialization (e.g., initialize only on first `tag(...)` call or behind a holder class) so the overhead is paid only when typed tail tagging is actually used. ########## tika-grpc-api/src/main/proto/org/apache/tika/grpc/v2/tika_v2.proto: ########## @@ -0,0 +1,90 @@ +// 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. + +// Experimental v2 parse surface. Fetcher/iterator management stays on the +// existing `tika.Tika` (v1) service; this service returns the typed Document +// contract instead of the v1 fields map. + +syntax = "proto3"; + +package org.apache.tika.grpc.v2; + +import "org/apache/tika/grpc/v2/document.proto"; + +option java_multiple_files = true; +option java_package = "org.apache.tika.grpc.v2"; +option java_outer_classname = "TikaV2Proto"; + +service TikaV2Service { + // Fetch via a previously saved fetcher and return a typed Document. + rpc FetchAndParse(FetchAndParseRequest) returns (FetchAndParseResponse) {} + // FetchAndParse as a server stream: one response, then the stream completes. + rpc FetchAndParseServerSideStreaming(FetchAndParseServerSideStreamingRequest) + returns (stream FetchAndParseServerSideStreamingResponse) {} + // One response for each request on the stream. + rpc FetchAndParseBiDirectionalStreaming(stream FetchAndParseBiDirectionalStreamingRequest) + returns (stream FetchAndParseBiDirectionalStreamingResponse) {} +} + +// Each RPC has its own request and response message so it can change on its own. Today +// the three requests carry the same fields under the same numbers, and so do the three +// responses. + +message FetchAndParseRequest { + // ID of a fetcher previously saved through the v1 Tika service. + string fetcher_id = 1; + string fetch_key = 2; + string additional_fetch_config_json = 3; + string emitter_id = 4; + // Optional JSON object to configure the ParseContext for this request. + string parse_context_json = 5; +} Review Comment: `emitter_id` is defined on all v2 requests, but the v2 server implementation shown in this PR does not read or forward it (it isn't represented in the v2 `FetchRequest` record nor passed into the shared fetch/parse path). If `emitter_id` is intentionally unsupported for v2 right now, document it as unused/reserved; otherwise, thread it through the implementation so the field has effect. ########## tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java: ########## @@ -638,6 +771,164 @@ 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)); + TikaV2ServiceGrpc.TikaV2ServiceStub v2 = TikaV2ServiceGrpc.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.FetchAndParseServerSideStreamingResponse> replies = + Collections.synchronizedList(new ArrayList<>()); + AtomicInteger errors = new AtomicInteger(); + AtomicInteger completions = new AtomicInteger(); + StreamObserver<org.apache.tika.grpc.v2.FetchAndParseServerSideStreamingResponse> observer = + new StreamObserver<>() { + @Override + public void onNext(org.apache.tika.grpc.v2.FetchAndParseServerSideStreamingResponse 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.FetchAndParseServerSideStreamingRequest + .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 results immediately after invoking an async server-streaming call. Even with `directExecutor()` and in-process transport, delivery/callback ordering can be brittle across gRPC versions and lead to intermittent failures. Prefer waiting on a `CountDownLatch`/`Future` for `onCompleted` (or for `replies.size() == 1`) before asserting counts. -- 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]
