This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 50d906acc6 TIKA-4833 -- fix Kafka iterator (#3048)
50d906acc6 is described below
commit 50d906acc65108ff580d11fbfb9c9b92a8cdc544
Author: Tim Allison <[email protected]>
AuthorDate: Sat Aug 22 16:01:19 2026 -0400
TIKA-4833 -- fix Kafka iterator (#3048)
---
CHANGES.txt | 9 +++
docs/modules/ROOT/pages/pipes/plugins/kafka.adoc | 74 +++++++++++++++++++---
.../tika/pipes/kafka/tests/TikaPipesKafkaTest.java | 50 +++++++++++++--
.../src/test/resources/kafka/plugins-template.json | 2 +-
.../pipes/iterator/kafka/KafkaPipesIterator.java | 46 ++++++++++++--
.../iterator/kafka/KafkaPipesIteratorConfig.java | 21 ++++++
6 files changed, 180 insertions(+), 22 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..e3b448e321 100644
--- a/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc
+++ b/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.
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..153a30224b 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;
@@ -92,7 +94,9 @@ public class TikaPipesKafkaTest {
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 class TikaPipesKafkaTest {
@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 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 +206,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 +243,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 +259,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;
}
}
