TIKA-4766 v2alpha1 WIP DO NOT MERGE

.github/workflows/buf.yml

@@ -0,0 +1,94 @@
+#
+# 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.
+#
+
+name: buf lint and breaking
+
+on:
+  pull_request:
+    branches: [ main ]
+    paths:
+      - '**/*.proto'
+      - '**/buf.yaml'
+      - '.github/workflows/buf.yml'
+  push:
+    branches: [ main ]
+    paths:
+      - '**/*.proto'
+      - '**/buf.yaml'
+      - '.github/workflows/buf.yml'
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+
+env:
+  BUF_VERSION: 1.50.0
+  # sha256 of buf-Linux-x86_64 from the release's sha256.txt; a download that does not
+  # match fails the job instead of running an unverified binary.
+  BUF_SHA256: 154ea883ce098eac4fa106ff9ee4e4964bb97f809dd8ec9c34a432b466ce1494
+
+jobs:
+  buf:
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    strategy:
+      fail-fast: false
+      matrix:
+        # tika-grpc holds the frozen v1 tika.proto, tika-grpc-api the v2alpha1 protos.
+        module: [ tika-grpc, tika-grpc-api ]
+    steps:
+      # actions/checkout v7.0.1, pinned by commit
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
+        with:
+          fetch-depth: 0
+      - name: install buf
+        run: |
+          curl -sSfL -o "$RUNNER_TEMP/buf" \
+            "https://github.com/bufbuild/buf/releases/download/v${BUF_VERSION}/buf-Linux-x86_64"
+          echo "${BUF_SHA256}  $RUNNER_TEMP/buf" | sha256sum -c -
+          chmod +x "$RUNNER_TEMP/buf"
+          echo "$RUNNER_TEMP" >> "$GITHUB_PATH"
+      - name: buf lint
+        working-directory: ${{ matrix.module }}
+        run: buf lint
+      # Pull requests compare with the base branch; pushes with the commit before the push.
+      # The rules come from the base's buf.yaml, so a change cannot loosen the check that
+      # judges it.
+      - name: buf breaking
+        env:
+          BASE: ${{ github.event_name == 'pull_request' && github.base_ref || github.event.before }}
+          MODULE: ${{ matrix.module }}
+        run: |
+          if [ "${{ github.event_name }}" = "pull_request" ]; then
+            git fetch --no-tags origin "$BASE"
+            BASE_SHA=$(git rev-parse FETCH_HEAD)
+          else
+            BASE_SHA="$BASE"
+          fi
+          if ! git cat-file -e "$BASE_SHA^{commit}" 2>/dev/null \
+              || [ -z "$(git ls-tree -r --name-only "$BASE_SHA" -- "$MODULE/src/main/proto" | grep '\.proto$')" ]; then
+            echo "$BASE has no protos under $MODULE yet; nothing to compare."
+            exit 0
+          fi
+          if git cat-file -e "$BASE_SHA:$MODULE/buf.yaml" 2>/dev/null; then
+            git show "$BASE_SHA:$MODULE/buf.yaml" > "$MODULE/buf.yaml"
+          fi
+          buf breaking "$MODULE" \
+            --against ".git#ref=$BASE_SHA,subdir=$MODULE/src/main/proto"

.skills/devs/metadata-schema/SKILL.md

@@ -88,6 +88,9 @@ under the full test classpath and asserts a byte-match — the definitive comple
   reflects the *registry*, not live code — regenerate before trusting it.
 - `MetadataKeyValidatorTest` — registry-driven CLOSED/OPEN/TEMPLATE/UNKNOWN classifier.
 - `MetadataCoverageTest` — fails if a scanned module's keys are neither in scope nor listed out-of-scope.
+- `MetadataTypeChangesTest` — a key's valueType/cardinality may differ from the committed baseline only
+  if `type-changes.json` declares it (key, from, to, jira). Clients consume these types (gRPC v2 hints,
+  typed /rmeta JSON), so a change also needs a CHANGES row. Refresh the baseline at release (README).
 
 Failures with stale `X-TIKA:`/underscore/`SHA256` keys usually mean *regenerate*, not edit code.
 

CHANGES.txt

@@ -22,6 +22,21 @@ Release 4.3.0 - ???
      for EKS IAM roles for service accounts (IRSA); the plugin now bundles the AWS STS
      module it needs (TIKA-5006).
 
+   * tika-grpc applies parse-context.timeout-limits and exception-reporting in the server
+     process too, so its backstop for a hung worker no longer falls back to the built-in
+     timeouts. A FetchAndParse call interrupted before the parse finished (server shutdown)
+     closes with UNAVAILABLE; a unary call used to fail with INTERNAL "no value received",
+     a server-streaming call completed with no reply, and a bidirectional stream went on
+     to throw on the closed call. Thanks to Kristian Rickert and Davide Polato (TIKA-5010).
+
+   * tika-grpc adds an experimental service, org.apache.tika.grpc.v2alpha1.TikaService, on
+     the same port as v1. It returns what /rmeta returns: one document per container and
+     embedded file, each with its content and every metadata value (v1 keeps one value per
+     key and merges embedded files into one map). Values are strings, with type and
+     cardinality hints from tika-metadata-schema. Protos are in the new tika-grpc-api
+     artifact. The package may change incompatibly until it becomes v2. Thanks to Kristian
+     Rickert and Davide Polato for the design work in #2961 (TIKA-4766).
+
    * The Kafka emitter reports success only once Kafka has acknowledged every record
      (per acks); a rejected or timed-out record fails the emit instead of being lost
      silently, and the producer is flushed and closed on shutdown. (TIKA-5008).

docs/modules/ROOT/pages/using-tika/grpc/index.adoc

@@ -25,15 +25,92 @@ The service definition lives in `tika-grpc/src/main/proto/tika.proto`. Clients r
 (`SaveFetcher`) and then submit `FetchAndParseRequest` messages, each returning a
 `FetchAndParseReply` with extracted metadata and content. Streaming variants
 (`FetchAndParseServerSideStreaming`, `FetchAndParseBiDirectionalStreaming`) return the same reply
-type.
+type. An experimental second service returns every document and every metadata value; see
+<<_v2alpha1,The v2alpha1 service>>.
 
 == Concurrency
 
 `FetchAndParse` and its streaming variants run on a pool of forked JVMs sized by
 `pipes.numClients` (default: derived from host cores, at most 4). A call that cannot get a fork
 within `pipes.maxWaitForClientMillis` (default 60s) returns the in-band reply status
 `CLIENT_UNAVAILABLE_WITHIN_MS` — at capacity, not failing. `pipes.useSharedServer: true` shares one
-JVM instead.
+JVM instead. (The v2alpha1 service answers that case with `RESOURCE_EXHAUSTED`.)
+
+== Message size
+
+gRPC clients reject replies larger than 4 MB by default, and a large document's text easily
+exceeds that. Raise the limit on the client channel, up to the server's ceiling,
+`pipes.maxIpcPayloadBytes` (100 MB by default):
+
+* Java: `ManagedChannelBuilder.forTarget(...).maxInboundMessageSize(100 * 1024 * 1024)`
+* Python: `grpc.insecure_channel(target, options=[("grpc.max_receive_message_length", 100 * 1024 * 1024)])`
+* Go: `grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(100 << 20))`
+* grpcurl: `-max-msg-sz 104857600`
+
+gRPC compression saves bandwidth but does not help with this limit, which applies to the
+decompressed size. `writeLimit` in the parse context caps how much text is extracted. The
+v2alpha1 `FetchAndParseStream` RPC sends one message per document, so an archive's attachments
+each count against the limit separately.
+
+[#_v2alpha1]
+== The v2alpha1 service (experimental)
+
+`org.apache.tika.grpc.v2alpha1.TikaService` runs on the same port as the v1 service. Its protos
+are in the `tika-grpc-api` artifact, so clients can depend on them without the server. The
+package may change incompatibly until it is renamed to `v2`.
+
+It returns what `/rmeta` returns. Each reply holds one `ParsedDocument` for the container and one
+for each embedded file, and each `ParsedDocument` carries:
+
+* `embedded_path`: empty for the container, otherwise the file's `tk:embedded-id-path`.
+* `content`: the extracted text, in the format named by `tk:content-handler-type`.
+* `metadata`: every key with all of its values. v1's `fields` map keeps one value per key and
+merges embedded files into one map.
+
+Metadata values are always strings, exactly as Tika stores them. Each key may also carry a type
+hint (`VALUE_TYPE_INTEGER`, `VALUE_TYPE_DATE`, ...) and a cardinality hint (`CARDINALITY_BAG`,
+`CARDINALITY_SEQ`, ...). Hints come from Tika's metadata key registry (`tika-metadata-schema`), and
+an absent hint means text. Hints can change between Tika versions, so treat them this way:
+
+* Read them on every response.
+* Fall back to the string when a value does not parse as its hinted type.
+* Never hard-code a key's type.
+
+Dates are written as `yyyy-MM-ddTHH:mm:ssZ` (UTC), `yyyy-MM-ddTHH:mm:ss` (no time zone was
+known), or `yyyy-MM-dd`.
+
+[cols="1,3",options="header"]
+|===
+|RPC |Behavior
+
+|`FetchAndParse`
+|One `FetchAndParseResponse` with every document.
+
+|`FetchAndParseStream`
+|One message per document, in the same order.
+
+|`FetchAndParseBidi`
+|Many requests on one stream, one response per request, matched by `request_id`. A request that
+fails gets a response with `error` set, and the stream continues.
+|===
+
+Failures that are not about the document close a unary or streaming call with a gRPC status:
+
+* `NOT_FOUND`: unknown fetcher.
+* `RESOURCE_EXHAUSTED`: no worker became free in time.
+* `UNAVAILABLE`: a worker or fetcher failed to start, or the server is shutting down.
+* `PERMISSION_DENIED`: per-request config was sent but is not allowed.
+* `INVALID_ARGUMENT`: bad `parse_context_json`.
+
+On `FetchAndParseBidi`, the same codes arrive in that request's `error` instead, and only server
+shutdown ends the stream. Everything about the document itself is a normal reply: `pipes_status`
+says what happened (`PARSE_SUCCESS`, `FETCH_EXCEPTION`, `TIMEOUT`, ...) and `message` says why. Each
+document's `tk:exception:*` metadata holds exceptions the parse recorded. How much exception detail
+callers see is set by
+xref:advanced/setting-limits.adoc#_exception_reporting[`parse-context.exception-reporting`].
+
+The v2alpha1 service goes through the same capability flags and request checks as v1. Fetchers are
+still managed through v1 (`SaveFetcher` and related RPCs).
 
 == Command-line options
 

pom.xml

@@ -54,6 +54,7 @@
     <module>tika-langdetect</module>
     <module>tika-pipes</module>
 
+    <module>tika-grpc-api</module>
     <module>tika-grpc</module>
     <module>tika-app</module>
     <module>tika-server</module>

tika-bom/pom.xml

@@ -387,6 +387,11 @@
         <artifactId>tika-async-cli</artifactId>
         <version>${revision}</version>
       </dependency>
+      <dependency>
+        <groupId>org.apache.tika</groupId>
+        <artifactId>tika-grpc-api</artifactId>
+        <version>${revision}</version>
+      </dependency>
       <dependency>
         <groupId>org.apache.tika</groupId>
         <artifactId>tika-httpclient-commons</artifactId>

tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java

@@ -89,14 +89,16 @@ static void setup() throws Exception {
     }
 
     /**
-     * The checked-in configs name the Docker image's plugins dir; the local server gets a copy
-     * pointed at the zip this module copies into target/plugins.
+     * The checked-in configs name the Docker image's plugins and fixtures dirs; the local server
+     * gets a copy pointed at target/plugins and the fixtures folder.
      */
     public static Path localConfig(String name) throws IOException {
         Path pluginsDir = Path.of("target/plugins").toAbsolutePath();
         String json = Files.readString(Path.of("src/test/resources", name));
         String local = json.replace("/var/cache/tika/plugins",
-                pluginsDir.toString().replace("\\", "\\\\"));
+                pluginsDir.toString().replace("\\", "\\\\"))
+                .replace(GOV_DOCS_FOLDER,
+                        TEST_FOLDER.getAbsolutePath().replace("\\", "\\\\"));
         Path out = Path.of("target", name).toAbsolutePath();
         Files.writeString(out, local);
         return out;

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

@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.tika.pipes.filesystem;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.ManagedChannel;
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+
+import org.apache.tika.grpc.v2alpha1.Cardinality;
+import org.apache.tika.grpc.v2alpha1.FetchAndParseBidiRequest;
+import org.apache.tika.grpc.v2alpha1.FetchAndParseBidiResponse;
+import org.apache.tika.grpc.v2alpha1.FetchAndParseRequest;
+import org.apache.tika.grpc.v2alpha1.FetchAndParseResponse;
+import org.apache.tika.grpc.v2alpha1.ParsedDocument;
+import org.apache.tika.grpc.v2alpha1.TikaServiceGrpc;
+import org.apache.tika.grpc.v2alpha1.Values;
+import org.apache.tika.pipes.ExternalTestBase;
+
+/** The v2alpha1 service on a live server, next to v1 on the same port. */
+@DisabledOnOs(value = OS.WINDOWS, disabledReason = "exec:exec classpath exceeds Windows CreateProcess command-line length limit")
+class V2Alpha1Test extends ExternalTestBase {
+
+    // declared in tika-config.json: a fetcher saved at runtime would not reach the forked
+    // workers under the default in-memory config store
+    private static final String FETCHER_ID = "fixtures";
+
+    @Test
+    void v2ParsesAndReportsFailuresInBand() throws Exception {
+        ManagedChannel channel = getManagedChannel();
+        try {
+            FetchAndParseResponse csv = TikaServiceGrpc.newBlockingStub(channel)
+                    .fetchAndParse(FetchAndParseRequest.newBuilder()
+                            .setFetcherId(FETCHER_ID).setFetchKey("sample.csv").build());
+            assertEquals("PARSE_SUCCESS", csv.getPipesStatus(), csv.getMessage());
+            ParsedDocument document = csv.getDocuments(0);
+            assertTrue(!document.getContent().isBlank(), "content");
+            assertTrue(document.containsMetadata("tk:content-handler-type"));
+            // a registry hint on the wire: every parser in the chain, as a BAG
+            Values parsedBy = document.getMetadataOrThrow("tk:parsed-by");
+            assertEquals(Cardinality.CARDINALITY_BAG, parsedBy.getCardinality());
+            assertTrue(parsedBy.getValuesCount() > 1, parsedBy.toString());
+
+            List<FetchAndParseBidiResponse> responses = Collections.synchronizedList(new ArrayList<>());
+            List<Throwable> errors = Collections.synchronizedList(new ArrayList<>());
+            CountDownLatch done = new CountDownLatch(1);
+            StreamObserver<FetchAndParseBidiRequest> requests = TikaServiceGrpc.newStub(channel)
+                    .fetchAndParseBidi(new StreamObserver<>() {
+                        @Override
+                        public void onNext(FetchAndParseBidiResponse response) {
+                            responses.add(response);
+                        }
+
+                        @Override
+                        public void onError(Throwable t) {
+                            errors.add(t);
+                            done.countDown();
+                        }
+
+                        @Override
+                        public void onCompleted() {
+                            done.countDown();
+                        }
+                    });
+            requests.onNext(bidi("html", FETCHER_ID, "sample.html"));
+            requests.onNext(bidi("missing-fetcher", "no-such-fetcher", "sample.html"));
+            requests.onNext(bidi("xml", FETCHER_ID, "sample.xml"));
+            requests.onCompleted();
+            assertTrue(done.await(2, TimeUnit.MINUTES), "bidi stream did not finish");
+
+            assertTrue(errors.isEmpty(), String.valueOf(errors));
+            Map<String, FetchAndParseBidiResponse> byId = new HashMap<>();
+            responses.forEach(r -> byId.put(r.getRequestId(), r));
+            assertEquals(3, byId.size());
+            assertEquals("PARSE_SUCCESS", byId.get("html").getResult().getPipesStatus());
+            assertEquals(Status.Code.NOT_FOUND.value(), byId.get("missing-fetcher").getError().getCode());
+            assertEquals("PARSE_SUCCESS", byId.get("xml").getResult().getPipesStatus());
+        } finally {
+            channel.shutdown();
+        }
+    }
+
+    private static FetchAndParseBidiRequest bidi(String id, String fetcherId, String fetchKey) {
+        return FetchAndParseBidiRequest.newBuilder()
+                .setRequestId(id).setFetcherId(fetcherId).setFetchKey(fetchKey).build();
+    }
+}

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

@@ -31,6 +31,13 @@
       "--add-opens=java.desktop/java.awt.font=ALL-UNNAMED"
     ]
   },
+  "fetchers": {
+    "fixtures": {
+      "file-system-fetcher": {
+        "basePath": "/tika/govdocs1"
+      }
+    }
+  },
   "grpc": {
     "allowComponentManagement": true
   }