diff --git a/CHANGES.txt b/CHANGES.txt index 7969560e697..bef8eef13b6 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -22,6 +22,10 @@ Release 4.3.0 - ??? for EKS IAM roles for service accounts (IRSA); the plugin now bundles the AWS STS module it needs (TIKA-5006). + * The Kafka emitter reports success only once Kafka has acknowledged every record + (per acks); a rejected or timed-out record fails the emit instead of being lost + silently, and the producer is flushed and closed on shutdown. (TIKA-5008). + * The JDBC emitter is thread-safe: concurrent emits (shared-server mode, several async emitter threads) shared one connection and prepared statement unguarded, mixing values across rows and committing each other's batches (TIKA-5007). diff --git a/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc b/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc index 34d695a7168..28c325b491a 100644 --- a/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc +++ b/docs/modules/ROOT/pages/pipes/plugins/kafka.adoc @@ -73,7 +73,8 @@ applies. The Default column below is therefore the Kafka client's default. |`lingerMs` |`5` -|Producer linger, ms. +|Producer linger, ms. Each emit waits for its records to be acknowledged, so a single-document +emit can wait up to this long for the batch to fill; keep it small. |`batchSize` |`16384` @@ -133,11 +134,12 @@ applies. The Default column below is therefore the Kafka client's default. |`transactionTimeoutMs` |`60000` -|Transaction timeout; only meaningful with `transactionalId`. +|Transaction timeout. Unused: the emitter does not use transactions. |`transactionalId` |_none_ -|Set to enable the transactional producer. +|Not supported; the emitter refuses to load with it set. It never opened a transaction, so +the producer rejected every send. |`clientId` |_none_ @@ -262,6 +264,6 @@ include::example$pipes-kafka-pipeline.json[] == Notes * 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`. +* An emit succeeds only once Kafka has acknowledged every record it sent, as defined by `acks`; a record Kafka rejects or does not acknowledge within `deliveryTimeoutMs` fails the emit (`EMIT_EXCEPTION`), so the caller can retry it. A batch from the async emitter is sent in full before waiting, one round trip per batch. +* `enableIdempotence: true` (with `acks: all`) keeps the producer's own retries from writing duplicates. A document the pipeline retries is sent again, so consumers should be idempotent on the record key. * 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-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitter.java b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitter.java index b4b95aacb27..f6153243839 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitter.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitter.java @@ -16,17 +16,22 @@ */ package org.apache.tika.pipes.emitter.kafka; +import java.io.Closeable; import java.io.IOException; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,6 +40,7 @@ import org.apache.tika.metadata.Metadata; import org.apache.tika.parser.ParseContext; import org.apache.tika.pipes.api.emitter.AbstractEmitter; +import org.apache.tika.pipes.api.emitter.EmitData; import org.apache.tika.plugins.ExtensionConfig; /** @@ -49,14 +55,14 @@ * "topic": "tika-output", * "bootstrapServers": "localhost:9092", * "acks": "all", - * "lingerMs": 5000 + * "lingerMs": 5 * } * } * } * } * */ -public class KafkaEmitter extends AbstractEmitter { +public class KafkaEmitter extends AbstractEmitter implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(KafkaEmitter.class); private static final ObjectMapper OM = new ObjectMapper(); @@ -71,7 +77,7 @@ public static KafkaEmitter build(ExtensionConfig extensionConfig) throws TikaCon return new KafkaEmitter(extensionConfig, config, producer); } - private KafkaEmitter(ExtensionConfig extensionConfig, KafkaEmitterConfig config, Producer producer) throws IOException { + KafkaEmitter(ExtensionConfig extensionConfig, KafkaEmitterConfig config, Producer producer) throws IOException { super(extensionConfig); this.config = config; this.producer = producer; @@ -132,6 +138,26 @@ public void emit(String emitKey, List metadataList, ParseContext parse if (metadataList == null || metadataList.isEmpty()) { throw new IOException("metadata list must not be null or of size 0"); } + List> sent = new ArrayList<>(); + send(emitKey, metadataList, sent); + awaitAcks(sent); + } + + /** Sends every record first, then waits once: one round trip for the batch, not per item. */ + @Override + public void emit(List emitData) throws IOException { + List> sent = new ArrayList<>(); + for (EmitData d : emitData) { + if (d.getMetadataList() == null || d.getMetadataList().isEmpty()) { + throw new IOException("metadata list must not be null or of size 0"); + } + send(d.getEmitKey(), d.getMetadataList(), sent); + } + awaitAcks(sent); + } + + private void send(String emitKey, List metadataList, + List> sent) throws IOException { for (Metadata metadata : metadataList) { LOGGER.trace("about to emit to target topic: ({}) path:({})", config.topic(), emitKey); @@ -144,7 +170,28 @@ public void emit(String emitKey, List metadataList, ParseContext parse fields.put(n, vals[0]); } - producer.send(new ProducerRecord<>(config.topic(), emitKey, OM.writeValueAsString(fields))); + sent.add(producer.send( + new ProducerRecord<>(config.topic(), emitKey, OM.writeValueAsString(fields)))); } } + + // send() only queues the record: success is the broker's ack, bounded by deliveryTimeoutMs + private static void awaitAcks(List> sent) throws IOException { + for (Future f : sent) { + try { + f.get(); + } catch (ExecutionException e) { + throw new IOException("Kafka did not accept the record", e.getCause()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("interrupted waiting for Kafka to accept the record", e); + } + } + } + + /** Flushes buffered records and releases the producer. */ + @Override + public void close() { + producer.close(); + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterConfig.java b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterConfig.java index 9c34d911bca..ecf7ce0bac3 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterConfig.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterConfig.java @@ -58,5 +58,10 @@ public void validate() throws TikaConfigException { if (bootstrapServers == null || bootstrapServers.isBlank()) { throw new TikaConfigException("'bootstrapServers' must not be empty"); } + // the emitter never opens a transaction, so the producer refuses every send + if (transactionalId != null) { + throw new TikaConfigException("'transactionalId' is not supported: the Kafka " + + "emitter does not use transactions"); + } } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterFactory.java index 1bff5425610..2ecced7b5f8 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterFactory.java @@ -36,7 +36,7 @@ * "topic": "my-topic", * "bootstrapServers": "localhost:9092", * "acks": "all", - * "lingerMs": 5000 + * "lingerMs": 5 * } * } * } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/test/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/test/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterTest.java new file mode 100644 index 00000000000..dfa1b946525 --- /dev/null +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/test/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitterTest.java @@ -0,0 +1,170 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tika.pipes.emitter.kafka; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.util.List; +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 org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.apache.tika.exception.TikaConfigException; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.pipes.api.emitter.EmitData; +import org.apache.tika.plugins.ExtensionConfig; + +/** An emit succeeds only once Kafka has accepted every record (TIKA-5008). */ +public class KafkaEmitterTest { + + private MockProducer producer; + private KafkaEmitter emitter; + private ExecutorService executor; + + @BeforeEach + public void setUp() throws Exception { + // autoComplete=false: each send stays pending until the test acks or fails it + producer = new MockProducer<>(false, null, new StringSerializer(), new StringSerializer()); + String json = "{\"topic\":\"t\",\"bootstrapServers\":\"localhost:9092\"}"; + emitter = new KafkaEmitter(new ExtensionConfig("k", "kafka-emitter", json), + KafkaEmitterConfig.load(json), producer); + executor = Executors.newSingleThreadExecutor(); + } + + @AfterEach + public void tearDown() { + executor.shutdownNow(); + } + + @Test + public void testEmitWaitsForAck() throws Exception { + Future emit = executor.submit(() -> { + emitter.emit("doc", List.of(metadata("a"), metadata("b")), new ParseContext()); + return null; + }); + awaitSends(2); + assertTrue(producer.completeNext()); + assertFalse(waitDone(emit), "reported success with a record still unacknowledged"); + assertTrue(producer.completeNext()); + emit.get(10, TimeUnit.SECONDS); + } + + @Test + public void testFailedSendFailsTheEmit() throws Exception { + Future emit = executor.submit(() -> { + emitter.emit("doc", List.of(metadata("a")), new ParseContext()); + return null; + }); + awaitSends(1); + assertTrue(producer.errorNext(new RuntimeException("broker said no"))); + ExecutionException e = assertThrows( + ExecutionException.class, () -> emit.get(10, TimeUnit.SECONDS)); + assertTrue(e.getCause() instanceof IOException, String.valueOf(e.getCause())); + } + + @Test + public void testBatchSendsAllBeforeWaiting() throws Exception { + List batch = List.of(emitData("d1", "a"), emitData("d2", "b")); + Future emit = executor.submit(() -> { + emitter.emit(batch); + return null; + }); + // both on the wire before the first ack: one round trip for the batch + awaitSends(2); + assertTrue(producer.completeNext()); + assertTrue(producer.completeNext()); + emit.get(10, TimeUnit.SECONDS); + } + + @Test + public void testTransactionalIdRefused() { + // never initTransactions(): the producer would refuse every send + assertThrows(TikaConfigException.class, () -> KafkaEmitterConfig.load( + "{\"topic\":\"t\",\"bootstrapServers\":\"h:9092\",\"transactionalId\":\"x\"}") + .validate()); + } + + @Test + public void testCloseClosesTheProducer() { + emitter.close(); + assertTrue(producer.closed()); + } + + private void awaitSends(int n) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (producer.history().size() < n) { + if (System.nanoTime() > deadline) { + throw new AssertionError("expected " + n + " sends, saw " + producer.history().size()); + } + Thread.sleep(10); + } + assertEquals(n, producer.history().size()); + } + + private static boolean waitDone(Future f) throws InterruptedException { + Thread.sleep(200); + return f.isDone(); + } + + private static EmitData emitData(String key, String v) { + return new EmitData() { + @Override + public String getEmitKey() { + return key; + } + + @Override + public List getMetadataList() { + return List.of(metadata(v)); + } + + @Override + public String getContainerStackTrace() { + return null; + } + + @Override + public long getEstimatedSizeBytes() { + return 0; + } + + @Override + public ParseContext getParseContext() { + return null; + } + }; + } + + private static Metadata metadata(String v) { + Metadata m = new Metadata(); + m.set("k", v); + return m; + } +} diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/test/resources/config-examples/kafka-emitter.json b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/test/resources/config-examples/kafka-emitter.json index 602266e8d32..1cdb507419c 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/test/resources/config-examples/kafka-emitter.json +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-kafka/src/test/resources/config-examples/kafka-emitter.json @@ -5,7 +5,7 @@ "topic": "tika-parsed-docs", "bootstrapServers": "kafka1.example.com:9092,kafka2.example.com:9092", "acks": "all", - "lingerMs": 5000, + "lingerMs": 5, "batchSize": 16384, "compressionType": "lz4", "enableIdempotence": true,