You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2026/08/22 10:00:54 UTC
(tika) 01/01: TIKA-4833 -- fix Kafka iterator
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 <ta...@apache.org>
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;
}
}