This is an automated email from the ASF dual-hosted git repository. tballison pushed a commit to branch TIKA-4833-kafka-iterator-assignment-race in repository https://gitbox.apache.org/repos/asf/tika.git
commit 44b9882dc950322db319da9fcade1692e222e894 Author: tallison <[email protected]> AuthorDate: Sat Aug 22 05:59:29 2026 -0400 TIKA-4833 -- fix Kafka iterator --- CHANGES.txt | 9 +++++ docs/modules/ROOT/pages/pipes/plugins/kafka.adoc | 18 ++++++++- .../tika/pipes/kafka/tests/TikaPipesKafkaTest.java | 40 +++++++++++++++++-- .../src/test/resources/kafka/plugins-template.json | 2 +- .../pipes/iterator/kafka/KafkaPipesIterator.java | 46 +++++++++++++++++++--- .../iterator/kafka/KafkaPipesIteratorConfig.java | 21 ++++++++++ 6 files changed, 123 insertions(+), 13 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 5c32de6a3d..eea5ad329b 100644 --- a/CHANGES.txt +++ b/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 diff --git a/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc b/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc index 99178e8827..fd7784dbc3 100644 --- a/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc +++ b/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc @@ -183,15 +183,29 @@ 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] diff --git a/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/java/org/apache/tika/pipes/kafka/tests/TikaPipesKafkaTest.java b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/java/org/apache/tika/pipes/kafka/tests/TikaPipesKafkaTest.java index 1f4a8c1d6a..572870db50 100644 --- a/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/java/org/apache/tika/pipes/kafka/tests/TikaPipesKafkaTest.java +++ b/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.Map; 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; @@ -117,10 +119,24 @@ public class TikaPipesKafkaTest { @Test public void testKafkaPipeIteratorAndEmitter(@TempDir Path pipesDirectory) throws Exception { + runPipeIteratorAndEmitter(pipesDirectory, 1000); + } + + /** + * A 1 ms poll always expires before the consumer's group join completes, which is the + * loaded-host case that made the iterator enqueue 0 files and report success (TIKA-4833). + * Deterministic here; on a fast box the 1000 ms variant above usually wins the race. + */ + @Test + public void testPollShorterThanGroupJoin(@TempDir Path pipesDirectory) throws Exception { + runPipeIteratorAndEmitter(pipesDirectory, 1); + } + + 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 +181,10 @@ public class TikaPipesKafkaTest { 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 +198,20 @@ public class TikaPipesKafkaTest { 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 +235,8 @@ public class TikaPipesKafkaTest { @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 +251,7 @@ public class TikaPipesKafkaTest { 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", diff --git a/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/kafka/plugins-template.json b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/kafka/plugins-template.json index 1954551646..f2d15245f8 100644 --- a/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/kafka/plugins-template.json +++ b/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" } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIterator.java b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIterator.java index 09c9f2b1a7..fd10c27dea 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIterator.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIterator.java @@ -69,7 +69,6 @@ public class KafkaPipesIterator extends PipesIteratorBase { 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 @@ public class KafkaPipesIterator extends PipesIteratorBase { 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 @@ public class KafkaPipesIterator extends PipesIteratorBase { 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); diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIteratorConfig.java b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIteratorConfig.java index 63342adbe3..e962a5c568 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIteratorConfig.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/iterator/kafka/KafkaPipesIteratorConfig.java @@ -47,7 +47,15 @@ public class KafkaPipesIteratorConfig extends PipesIteratorConfig { 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 class KafkaPipesIteratorConfig extends PipesIteratorConfig { 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 class KafkaPipesIteratorConfig extends PipesIteratorConfig { 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 class KafkaPipesIteratorConfig extends PipesIteratorConfig { result = 31 * result + pollDelayMs; result = 31 * result + emitMax; result = 31 * result + groupInitialRebalanceDelayMs; + result = 31 * result + assignmentTimeoutMs; + result = 31 * result + drainIdleMs; return result; } }
