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:53 UTC
(tika) branch TIKA-4833-kafka-iterator-assignment-race created (now 44b9882dc9)
This is an automated email from the ASF dual-hosted git repository.
tballison pushed a change to branch TIKA-4833-kafka-iterator-assignment-race
in repository https://gitbox.apache.org/repos/asf/tika.git
at 44b9882dc9 TIKA-4833 -- fix Kafka iterator
This branch includes the following new commits:
new 44b9882dc9 TIKA-4833 -- fix Kafka iterator
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(tika) 01/01: TIKA-4833 -- fix Kafka iterator
Posted by ta...@apache.org.
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;
}
}