TIKA-4833 -- fix Kafka iterator (#3048)

Author: tballison

CHANGES.txt

@@ -1,5 +1,14 @@
 Release 4.1.0 - unreleased
 
+   * The Kafka pipes iterator no longer stops at the first empty poll. A newly
+     subscribed consumer spends its first poll(s) joining the group and returns
+     empty even when the topic has a backlog, so the iterator could enqueue zero
+     files and report success. It now waits for a partition assignment (bounded
+     by the new assignmentTimeoutMs, default 30s) and requires a continuous quiet
+     window (drainIdleMs, default 1s) before concluding the topic is drained.
+     groupInitialRebalanceDelayMs is deprecated and no longer sent to the
+     consumer: it is a broker setting that Kafka has always ignored (TIKA-4833).
+
    * Pipes IPC: carry inline document bytes as a raw binary field beside the
      tuple in the request envelope -- never inside the tuple or its
      ParseContext -- and disable Smile's 7-bit binary encoding. Tuple JSON

docs/modules/ROOT/pages/pipes/plugins/kafka.adoc

@@ -21,6 +21,11 @@
 
 The Apache Kafka plugin (`tika-pipes-kafka`) provides an emitter (publishes parsed documents to a Kafka topic) and an iterator (consumes fetch requests from a Kafka topic).
 
+The two halves have different standing. The *emitter* is a plain producer and a good fit: Tika
+parses, results stream to a topic for downstream indexing. The *iterator* is a worked example
+whose suitability depends on your parse latency -- see <<iterator-caveats>> before building on
+it.
+
 [cols="2,1,3"]
 |===
 |Interface |Component name |Class
@@ -146,12 +151,47 @@ applies. The Default column below is therefore the Kafka client's default.
 [#kafka-iterator]
 == Kafka Iterator (`kafka-pipes-iterator`)
 
-Consumes fetch-request messages from a Kafka topic and emits one `FetchEmitTuple` per message. Useful for building event-driven pipelines where some upstream system pushes work to a queue.
-
-[source,json]
-----
-include::example$pipes-kafka-iterator.json[]
-----
+[WARNING]
+.Reference example -- check that your parse latency suits it
+====
+Kafka's consumer model assumes bounded, roughly uniform per-message processing time, so how
+well this component works depends on *how long your documents take to parse*. Short, predictable
+parses fit that assumption. Long-running parses do not: pairing Kafka with documents that take
+minutes to an hour -- OCR'd PDFs, large archives, anything near the one-hour default total-task
+timeout -- works against the grain of the offset model, and the caveats below stop being
+theoretical.
+
+It is kept as a worked example rather than a supported production integration. If your workload
+has a long latency tail, consider driving tika-server or tika-grpc from your own consumer, where
+you control acknowledgement and retry, and use the <<kafka-emitter,Kafka emitter>> to publish
+results.
+====
+
+[#iterator-caveats]
+=== Things to know before building on it
+
+*Head-of-line blocking, in proportion to your latency spread.* A partition is consumed in order
+by one member of the group. Even with `numClients` workers parsing in parallel, correct offset
+handling can only advance the commit watermark to the lowest un-acknowledged offset, so one slow
+document holds up its partition's progress. With short, uniform parses this is barely
+noticeable; with a long tail, one document can stall a partition for as long as it parses.
+This follows from Kafka's offset model rather than from a setting.
+
+*At-most-once delivery.* `enable.auto.commit` is left at Kafka's default of `true`, so offsets
+are committed on a timer as soon as records are *polled* -- before Tika has parsed or emitted
+them. A crash, OOM or failed emit in between loses those documents silently. Committing after a
+successful emit is not currently possible: the iterator pushes tuples onto a queue and receives
+no completion signal back.
+
+*It drains and exits; it does not stream.* The iterator enqueues what is on the topic and then
+finishes, because tika-pipes' iterator contract is finite. A quiet period ends the run (see
+`drainIdleMs`). It suits a periodic batch drain, not a long-running consumer.
+
+*Give the consumer room to join.* A stock Kafka broker applies
+`group.initial.rebalance.delay.ms` (default 3000) to the first member joining an *empty* group.
+A deployment that runs back-to-back, or keeps another member in the group, never pays this; one
+that starts cold pays it every run. The iterator waits for a partition assignment up to
+`assignmentTimeoutMs` rather than mistaking a not-yet-assigned consumer for an empty topic.
 
 === Configuration
 
@@ -183,21 +223,35 @@ In addition to the required `fetcherId` / `emitterId` (see xref:pipes/iterators.
 
 |`pollDelayMs`
 |`100`
-|Sleep between `poll()` calls when the topic is idle.
+|Timeout passed to each `poll()` call.
 
 |`emitMax`
 |`-1`
 |Maximum tuples to emit. `-1` means unbounded.
 
+|`assignmentTimeoutMs`
+|`30000`
+|How long to wait for the consumer to be assigned a partition before failing. A newly
+subscribed consumer returns empty polls while it joins the group; the iterator waits for an
+assignment so it cannot mistake that for an empty topic.
+
+|`drainIdleMs`
+|`1000`
+|How long the topic must stay quiet (no records, after assignment) before it is treated as
+drained and the iterator finishes. A duration rather than a poll count, so it holds however
+short `pollDelayMs` is.
+
 |`groupInitialRebalanceDelayMs`
 |`3000`
-|Initial rebalance delay for the consumer group.
+|*Deprecated and ignored.* This is a broker setting, not a consumer one, so Kafka never
+applied it. Use `assignmentTimeoutMs` instead. Still accepted so existing configs start;
+scheduled for removal.
 |===
 
 [#kafka-pipeline]
 == Complete Pipeline Example
 
-A Kafka iterator (consuming fetch requests), a filesystem fetcher, and a Kafka emitter (publishing parsed results) — the stream-processing shape.
+A Kafka iterator (consuming fetch requests), a filesystem fetcher, and a Kafka emitter (publishing parsed results). This end-to-end shape is the reference example; see <<iterator-caveats>> before relying on the iterator half.
 
 [source,json]
 ----
@@ -210,4 +264,4 @@ include::example$pipes-kafka-pipeline.json[]
 * The Kafka plugin uses the official `kafka-clients` SDK.
 * The emitter is fire-and-forget at the Tika level; durability is determined by Kafka's `acks` and broker replication factor, not by Tika.
 * For exactly-once semantics, set `enableIdempotence: true` (and ensure `acks: all`); for transactional semantics, also set `transactionalId`.
-* The iterator's `groupId` controls partition assignment. Set it explicitly in production — without one, the consumer receives a transient assignment that resets on restart.
+* The iterator's `groupId` controls partition assignment. Set it explicitly — without one, the consumer receives a transient assignment that resets on restart.

tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/java/org/apache/tika/pipes/kafka/tests/TikaPipesKafkaTest.java

@@ -33,8 +33,10 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.core.type.TypeReference;
@@ -92,7 +94,9 @@ public static void setUp() {
     private final ObjectMapper objectMapper = new ObjectMapper();
     private final Set<String> waitingFor = new HashSet<>();
     // https://java.testcontainers.org/modules/kafka/#using-orgtestcontainerskafkaconfluentkafkacontainer
-    ConfluentKafkaContainer kafka = new ConfluentKafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.4.0"));
+    private static final DockerImageName KAFKA_IMAGE =
+            DockerImageName.parse("confluentinc/cp-kafka:7.4.0");
+    ConfluentKafkaContainer kafka = new ConfluentKafkaContainer(KAFKA_IMAGE);
 
     private void createTestFiles(Path testFileFolderPath) throws Exception {
         Files.createDirectories(testFileFolderPath);
@@ -117,10 +121,30 @@ public void after() {
 
     @Test
     public void testKafkaPipeIteratorAndEmitter(@TempDir Path pipesDirectory) throws Exception {
+        runPipeIteratorAndEmitter(pipesDirectory, 1000);
+    }
+
+    /**
+     * Testcontainers pins the broker's group.initial.rebalance.delay.ms to 0; Kafka's own
+     * default is 3000, and tika-pipes leaves the group empty between runs, so a real
+     * deployment pays that delay on every run. Restore the stock value here: with the default
+     * 100 ms pollDelayMs the iterator used to give up ~30x too early, enqueue 0 files and
+     * report success (TIKA-4833). Without this the suite cannot see the production case at all.
+     */
+    @Test
+    public void testStockBrokerRebalanceDelay(@TempDir Path pipesDirectory) throws Exception {
+        kafka.close();
+        kafka = new ConfluentKafkaContainer(KAFKA_IMAGE)
+                .withEnv("KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS", "3000");
+        kafka.start();
+        runPipeIteratorAndEmitter(pipesDirectory, 100);
+    }
+
+    private void runPipeIteratorAndEmitter(Path pipesDirectory, int pollDelayMs) throws Exception {
         Path testFileFolderPath = pipesDirectory.resolve("test-files");
         createTestFiles(testFileFolderPath);
 
-        Path tikaConfigPath = getTikaConfig(pipesDirectory, testFileFolderPath);
+        Path tikaConfigPath = getTikaConfig(pipesDirectory, testFileFolderPath, pollDelayMs);
 
         Properties consumerProps = new Properties();
         consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
@@ -165,7 +189,10 @@ public void testKafkaPipeIteratorAndEmitter(@TempDir Path pipesDirectory) throws
             LOG.info("Producer is now complete - sent {}.", numSent);
         }
 
-        es.execute(() -> {
+        // Keep the Future: a bare execute() sent any TikaCLI failure to the thread's uncaught
+        // handler, so a broken pipeline surfaced only as the generic "timed out waiting for the
+        // emitted docs" below, hiding the real cause (TIKA-4833).
+        Future<?> pipesRun = es.submit(() -> {
             try {
                 TikaCLI.main(new String[]{"-a", "-c", tikaConfigPath.toAbsolutePath().toString()});
             } catch (Exception e) {
@@ -179,9 +206,20 @@ public void testKafkaPipeIteratorAndEmitter(@TempDir Path pipesDirectory) throws
 
         long startNanos = System.nanoTime();
         while (!waitingFor.isEmpty()) {
+            // Surface a tika-pipes failure as itself rather than as a timeout.
+            if (pipesRun.isDone()) {
+                try {
+                    pipesRun.get();
+                } catch (ExecutionException e) {
+                    throw new AssertionError(
+                            "tika-pipes failed before emitting all docs; still waiting for " +
+                                    waitingFor.size() + " of " + numDocs, e.getCause());
+                }
+            }
             assertFalse(TimeUnit.NANOSECONDS.toMinutes(System.nanoTime() - startNanos) > WAIT_FOR_EMITTED_DOCS_TIMEOUT_MINUTES,
                     "Timed out after " + WAIT_FOR_EMITTED_DOCS_TIMEOUT_MINUTES +
-                            " minutes waiting for the emitted docs");
+                            " minutes waiting for the emitted docs; still waiting for " +
+                            waitingFor.size() + " of " + numDocs);
             try {
                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                 for (ConsumerRecord<String, String> record : records) {
@@ -205,7 +243,8 @@ public void testKafkaPipeIteratorAndEmitter(@TempDir Path pipesDirectory) throws
 
 
     @NotNull
-    private Path getTikaConfig(Path pipesDirectory, Path testFileFolderPath) throws Exception {
+    private Path getTikaConfig(Path pipesDirectory, Path testFileFolderPath, int pollDelayMs)
+            throws Exception {
         Path tikaConfig = pipesDirectory.resolve("tika-config.json");
 
         Path log4jPropFile = pipesDirectory.resolve("log4j2.xml");
@@ -220,6 +259,7 @@ private Path getTikaConfig(Path pipesDirectory, Path testFileFolderPath) throws
         replacements.put("BOOTSTRAP_SERVERS", kafka.getBootstrapServers());
         replacements.put("FETCHER_BASE_PATH", testFileFolderPath);
         replacements.put("PARSE_MODE", ParseMode.RMETA.name());
+        replacements.put("POLL_DELAY_MS", pollDelayMs);
         replacements.put("LOG4J_JVM_ARG", "-Dlog4j.configurationFile=" + log4jPropFile.toAbsolutePath());
 
         JsonConfigHelper.writeConfigFromResource("/kafka/plugins-template.json",

tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/kafka/plugins-template.json

@@ -83,7 +83,7 @@
       "bootstrapServers": "BOOTSTRAP_SERVERS",
       "groupId": "grpid",
       "autoOffsetReset": "earliest",
-      "pollDelayMs": 1000,
+      "pollDelayMs": "POLL_DELAY_MS",
       "fetcherId": "fsf",
       "emitterId": "ke"
     }

tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIterator.java

@@ -69,7 +69,6 @@ private void configure() throws IOException, TikaConfigException {
                 serializerClass(config.getValueSerializer(), StringDeserializer.class));
         safePut(props, ConsumerConfig.GROUP_ID_CONFIG, config.getGroupId());
         safePut(props, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, config.getAutoOffsetReset());
-        safePut(props, "group.initial.rebalance.delay.ms", config.getGroupInitialRebalanceDelayMs());
 
         consumer = new KafkaConsumer<>(props);
         consumer.subscribe(Arrays.asList(config.getTopic()));
@@ -103,10 +102,29 @@ protected void enqueue() throws InterruptedException, TimeoutException {
         long start = System.currentTimeMillis();
         int count = 0;
         int emitMax = config.getEmitMax();
-        ConsumerRecords<String, String> records;
-
-        do {
-            records = consumer.poll(Duration.ofMillis(config.getPollDelayMs()));
+        boolean assigned = false;
+        long assignmentDeadline = start + config.getAssignmentTimeoutMs();
+        long idleSince = 0;
+
+        while (true) {
+            ConsumerRecords<String, String> records =
+                    consumer.poll(Duration.ofMillis(config.getPollDelayMs()));
+            // A freshly subscribed consumer spends its first poll(s) joining the group, and
+            // those return empty even when the topic has a backlog. Treating that as "drained"
+            // silently enqueued nothing and reported success (TIKA-4833), so wait for a
+            // partition assignment before an empty poll is allowed to end the loop.
+            if (!assigned) {
+                assigned = !consumer.assignment().isEmpty();
+                if (!assigned) {
+                    if (System.currentTimeMillis() > assignmentDeadline) {
+                        throw new TimeoutException(
+                                "Kafka consumer was not assigned a partition of topic '" +
+                                        config.getTopic() + "' within " +
+                                        config.getAssignmentTimeoutMs() + " ms");
+                    }
+                    continue;
+                }
+            }
             for (ConsumerRecord<String, String> r : records) {
                 long elapsed = System.currentTimeMillis() - start;
                 if (LOGGER.isDebugEnabled()) {
@@ -118,7 +136,23 @@ protected void enqueue() throws InterruptedException, TimeoutException {
                         FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT));
                 ++count;
             }
-        } while ((emitMax < 0 || count < emitMax) && !records.isEmpty());
+            if (emitMax >= 0 && count >= emitMax) {
+                break;
+            }
+            // A single empty poll mid-drain doesn't mean the topic is exhausted -- records may
+            // simply not have arrived yet -- so require a continuous quiet window. This is a
+            // duration, not a poll count, so it holds however short pollDelayMs is.
+            if (records.isEmpty()) {
+                long now = System.currentTimeMillis();
+                if (idleSince == 0) {
+                    idleSince = now;
+                } else if (now - idleSince >= config.getDrainIdleMs()) {
+                    break;
+                }
+            } else {
+                idleSince = 0;
+            }
+        }
 
         long elapsed = System.currentTimeMillis() - start;
         LOGGER.info("Finished enqueuing {} files in {} ms", count, elapsed);

tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIteratorConfig.java

@@ -47,7 +47,15 @@ public static KafkaPipesIteratorConfig load(final String json)
     private String autoOffsetReset = "earliest";
     private int pollDelayMs = 100;
     private int emitMax = -1;
+    /**
+     * @deprecated inert -- this is a broker setting, never a consumer one, so Kafka has always
+     * ignored it ("supplied but not used yet"). Kept only so existing configs still start;
+     * remove in 4.1.0. Use assignmentTimeoutMs to bound waiting for a partition assignment.
+     */
+    @Deprecated
     private int groupInitialRebalanceDelayMs = 3000;
+    private int assignmentTimeoutMs = 30000;
+    private int drainIdleMs = 1000;
 
     public String getTopic() {
         return topic;
@@ -81,10 +89,19 @@ public int getEmitMax() {
         return emitMax;
     }
 
+    @Deprecated
     public int getGroupInitialRebalanceDelayMs() {
         return groupInitialRebalanceDelayMs;
     }
 
+    public int getAssignmentTimeoutMs() {
+        return assignmentTimeoutMs;
+    }
+
+    public int getDrainIdleMs() {
+        return drainIdleMs;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (!(o instanceof KafkaPipesIteratorConfig that)) {
@@ -96,6 +113,8 @@ public boolean equals(Object o) {
         return pollDelayMs == that.pollDelayMs &&
                 emitMax == that.emitMax &&
                 groupInitialRebalanceDelayMs == that.groupInitialRebalanceDelayMs &&
+                assignmentTimeoutMs == that.assignmentTimeoutMs &&
+                drainIdleMs == that.drainIdleMs &&
                 Objects.equals(topic, that.topic) &&
                 Objects.equals(bootstrapServers, that.bootstrapServers) &&
                 Objects.equals(keySerializer, that.keySerializer) &&
@@ -116,6 +135,8 @@ public int hashCode() {
         result = 31 * result + pollDelayMs;
         result = 31 * result + emitMax;
         result = 31 * result + groupInitialRebalanceDelayMs;
+        result = 31 * result + assignmentTimeoutMs;
+        result = 31 * result + drainIdleMs;
         return result;
     }
 }