This is an automated email from the ASF dual-hosted git repository.
tallison 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 04d00a0ae TIKA-3820 add tika pipes kafka support (#616)
04d00a0ae is described below
commit 04d00a0aecd0dc4c0f4250fa972d407face06bb5
Author: Nicholas DiPiazza <[email protected]>
AuthorDate: Mon Jul 25 11:52:34 2022 -0500
TIKA-3820 add tika pipes kafka support (#616)
* wip - add kafka support
* wip - add kafka support
* finish setting up intiial kafka commit - tests pass.
* fix readme
* remove dead file, rename good test
* remove stuff from earlier test attempts
* deleted wrong file previously.
* bump version after merge.
* add an emit max in case the documents keep coming into the pipe iterator
forever.
* clean up javadoc.
* fix a few issues with the test.
* remove commented out line.
* no need to create the topics manually after i fixed issue with
conflicting group ids
* remove create topic code.
* add javadoc prefix
* looks like a refactor renamed a couple files
* improve messages
* fix poor message in javadoc.
* apply change suggested in https://stackoverflow.com/a/73024959/1174024
* clarify
* fix message
* fix annoying IDE warnings
---
tika-bom/pom.xml | 10 +
tika-integration-tests/pom.xml | 1 +
.../tika-pipes-kafka-integration-tests/README.md | 7 +
.../pom.xml | 50 ++--
.../tika/pipes/kafka/tests/TikaPipesKafkaTest.java | 215 ++++++++++++++++
.../src/test/resources/log4j2.xml | 32 +++
.../resources/pipes-fork-server-custom-log4j2.xml | 32 +++
.../src/test/resources/tika-config-kafka.xml | 118 +++++++++
.../pipes/solr/tests/TikaPipesSolrTestBase.java | 1 -
tika-parent/pom.xml | 1 +
tika-pipes/tika-emitters/pom.xml | 1 +
.../tika-emitters/tika-emitter-kafka/pom.xml | 132 ++++++++++
.../tika/pipes/emitter/kafka/KafkaEmitter.java | 284 +++++++++++++++++++++
tika-pipes/tika-pipes-iterators/pom.xml | 1 +
.../tika-pipes-iterator-kafka/pom.xml | 125 +++++++++
.../pipesiterator/kafka/KafkaPipesIterator.java | 174 +++++++++++++
.../kafka/TestKafkaPipesIterator.java | 98 +++++++
.../src/test/resources/log4j2.xml | 32 +++
18 files changed, 1294 insertions(+), 20 deletions(-)
diff --git a/tika-bom/pom.xml b/tika-bom/pom.xml
index 5226c95fd..c969b7654 100644
--- a/tika-bom/pom.xml
+++ b/tika-bom/pom.xml
@@ -318,6 +318,11 @@
<artifactId>tika-emitter-s3</artifactId>
<version>2.4.2-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-emitter-kafka</artifactId>
+ <version>2.4.2-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.tika</groupId>
<artifactId>tika-emitter-solr</artifactId>
@@ -360,6 +365,11 @@
<artifactId>tika-pipes-iterator-s3</artifactId>
<version>2.4.2-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-pipes-iterator-kafka</artifactId>
+ <version>2.4.2-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>org.apache.tika</groupId>
<artifactId>tika-pipes-iterator-solr</artifactId>
diff --git a/tika-integration-tests/pom.xml b/tika-integration-tests/pom.xml
index aecdef34c..f358458d9 100644
--- a/tika-integration-tests/pom.xml
+++ b/tika-integration-tests/pom.xml
@@ -36,6 +36,7 @@
<module>tika-pipes-opensearch-integration-tests</module>
<module>tika-pipes-s3-integration-tests</module>
<module>tika-resource-loading-tests</module>
+ <module>tika-pipes-kafka-integration-tests</module>
</modules>
<dependencies>
diff --git
a/tika-integration-tests/tika-pipes-kafka-integration-tests/README.md
b/tika-integration-tests/tika-pipes-kafka-integration-tests/README.md
new file mode 100644
index 000000000..2bc76c8fb
--- /dev/null
+++ b/tika-integration-tests/tika-pipes-kafka-integration-tests/README.md
@@ -0,0 +1,7 @@
+# Tika Pipes - Kafka - Integration tests
+
+The following test uses TestContainers to test the Tika Pipes Kafka modules:
+
+* Kafka Pipe Iterator
+* Kafka Emitter
+
diff --git a/tika-integration-tests/pom.xml
b/tika-integration-tests/tika-pipes-kafka-integration-tests/pom.xml
similarity index 57%
copy from tika-integration-tests/pom.xml
copy to tika-integration-tests/tika-pipes-kafka-integration-tests/pom.xml
index aecdef34c..e7447c98b 100644
--- a/tika-integration-tests/pom.xml
+++ b/tika-integration-tests/tika-pipes-kafka-integration-tests/pom.xml
@@ -17,28 +17,30 @@
specific language governing permissions and limitations
under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
https://maven.apache.org/xsd/maven-4.0.0.xsd";>
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<parent>
+ <artifactId>tika-integration-tests</artifactId>
<groupId>org.apache.tika</groupId>
- <artifactId>tika-parent</artifactId>
<version>2.4.2-SNAPSHOT</version>
- <relativePath>../tika-parent/pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>tika-integration-tests</artifactId>
- <name>Apache Tika integration tests</name>
-
- <packaging>pom</packaging>
-
- <modules>
- <module>tika-pipes-solr-integration-tests</module>
- <module>tika-pipes-opensearch-integration-tests</module>
- <module>tika-pipes-s3-integration-tests</module>
- <module>tika-resource-loading-tests</module>
- </modules>
+ <artifactId>tika-pipes-kafka-integration-tests</artifactId>
+ <name>Apache Tika Kafka pipes integration tests</name>
<dependencies>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <version>${test.containers.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>kafka</artifactId>
+ <version>${test.containers.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>tika-core</artifactId>
@@ -47,15 +49,25 @@
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>tika-serialization</artifactId>
+ <artifactId>tika-pipes-iterator-kafka</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-emitter-kafka</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-app</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
- <!-- after we migrate everything to junit5, we can get rid of this -->
<dependency>
- <groupId>org.junit.vintage</groupId>
- <artifactId>junit-vintage-engine</artifactId>
- <version>${junit5.version}</version>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
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
new file mode 100644
index 000000000..18a3cbf45
--- /dev/null
+++
b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/java/org/apache/tika/pipes/kafka/tests/TikaPipesKafkaTest.java
@@ -0,0 +1,215 @@
+/*
+ * 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.kafka.tests;
+
+import java.io.File;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Stopwatch;
+import org.apache.commons.io.FilenameUtils;
+import org.apache.commons.io.IOUtils;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.tika.cli.TikaCLI;
+import org.apache.tika.pipes.HandlerConfig;
+import org.jetbrains.annotations.NotNull;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.shaded.org.apache.commons.io.FileUtils;
+import org.testcontainers.utility.DockerImageName;
+
+/**
+ * Test will emit some documents into a Kafka "pipe_iterator_topic", then
kafka pipe iterator will
+ * poll those documents and send them to tika pipes. Tika pipes will then use
a file fetcher to fetch/parse, then
+ * the kafka emitter will send the now-parsed output to the "emitter_topic".
+ * Will then wait for the messages to come from the emitter and assert they
are correct.
+ */
+public class TikaPipesKafkaTest {
+ private static final Logger LOG =
LoggerFactory.getLogger(TikaPipesKafkaTest.class);
+
+ public static final String PIPE_ITERATOR_TOPIC = "pipe_iterator_topic";
+ public static final String EMITTER_TOPIC = "emitter_topic";
+ /**
+ * Wait up to this many minutes before you give up waiting for the emitted
documents to poll from the
+ * emitter_topic and fail the test.
+ */
+ public static final int WAIT_FOR_EMITTED_DOCS_TIMEOUT_MINUTES = 2;
+ private final int numDocs = 42;
+
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ private final File testFileFolder = new File("target", "test-files");
+
+ private final Set<String> waitingFor = new HashSet<>();
+
+ private void createTestFiles() throws Exception {
+ if (testFileFolder.mkdirs()) {
+ LOG.info("Created test folder: {}", testFileFolder);
+ }
+ for (int i = 0; i < numDocs; ++i) {
+ String nextFileName = "test-" + i + ".html";
+ FileUtils.writeStringToFile(new File(testFileFolder, nextFileName),
+ "<html><body>body-" + i + "</body></html>",
StandardCharsets.UTF_8);
+ waitingFor.add(nextFileName);
+ }
+ }
+
+ KafkaContainer kafka = new
KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1"));
+
+ @Before
+ public void before() {
+ kafka.start();
+ }
+
+ @After
+ public void after() {
+ kafka.close();
+ }
+
+ @Test
+ public void testKafkaPipeIteratorAndEmitter()
+ throws Exception {
+ createTestFiles();
+ File tikaConfigFile = new File("target", "ta.xml");
+ File log4jPropFile = new File("target", "tmp-log4j2.xml");
+ try (InputStream is = this.getClass()
+ .getResourceAsStream("/pipes-fork-server-custom-log4j2.xml")) {
+ assert is != null;
+ FileUtils.copyInputStreamToFile(is, log4jPropFile);
+ }
+ String tikaConfigTemplateXml;
+ try (InputStream is = this.getClass()
+ .getResourceAsStream("/tika-config-kafka.xml")) {
+ assert is != null;
+ tikaConfigTemplateXml = IOUtils.toString(is,
StandardCharsets.UTF_8);
+ }
+
+ Properties consumerProps = new Properties();
+ consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafka.getBootstrapServers());
+ consumerProps.put("group.id", UUID.randomUUID().toString());
+ consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
+ consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+ Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafka.getBootstrapServers());
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
+
+ KafkaConsumer<String, String> consumer = new
KafkaConsumer<>(consumerProps);
+ LOG.info("Listening to EMITTER_TOPIC={}", EMITTER_TOPIC);
+ consumer.subscribe(Collections.singletonList(EMITTER_TOPIC));
+
+ ExecutorService es = Executors.newCachedThreadPool();
+
+ try (KafkaProducer<String, String> producer = new
KafkaProducer<>(producerProps)) {
+ int numSent = 0;
+ for (int i = 0; i < numDocs; ++i) {
+ File nextFile = new File(testFileFolder, "test-" + i +
".html");
+ Map<String, Object> meta = new HashMap<>();
+ meta.put("name", nextFile.getName());
+ meta.put("path", nextFile.getAbsolutePath());
+ meta.put("totalSpace", nextFile.getTotalSpace());
+ try {
+ producer.send(new ProducerRecord<>(PIPE_ITERATOR_TOPIC,
+ nextFile.getAbsolutePath(),
+ objectMapper.writeValueAsString(meta))).get();
+ LOG.info("Sent fetch request : {}",
nextFile.getAbsolutePath());
+ ++numSent;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ LOG.info("Producer is now complete - sent {}.", numSent);
+ }
+
+ es.execute(() -> {
+ try {
+ String tikaConfigXml =
+ createTikaConfigXml(tikaConfigFile, log4jPropFile,
tikaConfigTemplateXml
+ );
+
+ FileUtils.writeStringToFile(tikaConfigFile, tikaConfigXml,
StandardCharsets.UTF_8);
+ TikaCLI.main(new String[] {"-a", "--config=" +
tikaConfigFile.getAbsolutePath()});
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ LOG.info("Tika pipes have been started. See if we can pull the
response messages from the EMITTER_TOPIC={}", EMITTER_TOPIC);
+
+ Stopwatch stopwatch = Stopwatch.createStarted();
+ while (!waitingFor.isEmpty()) {
+ Assert.assertFalse("Timed out after " +
WAIT_FOR_EMITTED_DOCS_TIMEOUT_MINUTES + " minutes waiting for the emitted docs",
+ stopwatch.elapsed(TimeUnit.MINUTES) >
WAIT_FOR_EMITTED_DOCS_TIMEOUT_MINUTES);
+ try {
+ ConsumerRecords<String, String> records =
consumer.poll(Duration.ofSeconds(1));
+ for (ConsumerRecord<String, String> record : records) {
+ String val = record.value();
+ Map<String, Object> valMap = objectMapper.readValue(val,
new TypeReference<Map<String, Object>>() {});
+ waitingFor.remove(FilenameUtils.getName(record.key()));
+ Assert.assertNotNull(valMap.get("content_s"));
+ Assert.assertNotNull(valMap.get("mime_s"));
+ Assert.assertNotNull(valMap.get("length_i"));
+ LOG.info("Received message key={}, offset={}",
record.key(), record.offset());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ LOG.info("Done");
+ }
+
+ @NotNull
+ private String createTikaConfigXml(File tikaConfigFile, File log4jPropFile,
+ String tikaConfigTemplateXml) {
+ return tikaConfigTemplateXml.replace("{TIKA_CONFIG}",
tikaConfigFile.getAbsolutePath())
+ .replace("{LOG4J_PROPERTIES_FILE}",
log4jPropFile.getAbsolutePath())
+ .replace("{PATH_TO_DOCS}",
testFileFolder.getAbsolutePath())
+ .replace("{PARSE_MODE}",
HandlerConfig.PARSE_MODE.RMETA.name())
+ .replace("{PIPE_ITERATOR_TOPIC}", PIPE_ITERATOR_TOPIC)
+ .replace("{EMITTER_TOPIC}", EMITTER_TOPIC)
+ .replace("{BOOTSTRAP_SERVERS}",
kafka.getBootstrapServers());
+ }
+}
diff --git
a/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/log4j2.xml
b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/log4j2.xml
new file mode 100644
index 000000000..c88e66e99
--- /dev/null
+++
b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/log4j2.xml
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no" ?>
+
+<!--
+ 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.
+-->
+<Configuration status="WARN">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_ERR">
+ <PatternLayout pattern="%-5p [%t] %d{HH:mm:ss,SSS} %c %m%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="info">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>
\ No newline at end of file
diff --git
a/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/pipes-fork-server-custom-log4j2.xml
b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/pipes-fork-server-custom-log4j2.xml
new file mode 100644
index 000000000..5f946e6e5
--- /dev/null
+++
b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/pipes-fork-server-custom-log4j2.xml
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<Configuration status="INFO">
+ <Appenders>
+ <Console name="console" target="SYSTEM_ERR">
+ <PatternLayout
+ pattern="%-5p [%t] %d{HH:mm:ss,SSS} %c %m%n" />
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="info" additivity="false">
+ <AppenderRef ref="console" />
+ </Root>
+ </Loggers>
+</Configuration>
diff --git
a/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/tika-config-kafka.xml
b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/tika-config-kafka.xml
new file mode 100644
index 000000000..f15faf262
--- /dev/null
+++
b/tika-integration-tests/tika-pipes-kafka-integration-tests/src/test/resources/tika-config-kafka.xml
@@ -0,0 +1,118 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+ 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.
+-->
+<properties>
+ <parsers>
+ <parser class="org.apache.tika.parser.DefaultParser">
+ <parser-exclude
class="org.apache.tika.parser.ocr.TesseractOCRParser"/>
+ <parser-exclude class="org.apache.tika.parser.pdf.PDFParser"/>
+ <parser-exclude
class="org.apache.tika.parser.microsoft.ooxml.OOXMLParser"/>
+ <parser-exclude
class="org.apache.tika.parser.microsoft.OfficeParser"/>
+ </parser>
+ <parser class="org.apache.tika.parser.pdf.PDFParser">
+ <params>
+ <param name="extractActions" type="bool">true</param>
+ <param name="checkExtractAccessPermissions"
type="bool">true</param>
+ </params>
+ </parser>
+ <parser class="org.apache.tika.parser.microsoft.ooxml.OOXMLParser">
+ <params>
+ <param name="includeDeletedContent" type="bool">true</param>
+ <param name="includeMoveFromContent" type="bool">true</param>
+ <param name="extractMacros" type="bool">true</param>
+ </params>
+ </parser>
+ <parser class="org.apache.tika.parser.microsoft.OfficeParser">
+ <params>
+ <param name="extractMacros" type="bool">true</param>
+ </params>
+ </parser>
+ </parsers>
+ <metadataFilters>
+ <!-- depending on the file format, some dates do not have a timezone.
This
+ filter arbitrarily assumes dates have a UTC timezone and will
format all
+ dates as yyyy-MM-dd'T'HH:mm:ss'Z' whether or not they actually
have a timezone.
+ -->
+ <metadataFilter
class="org.apache.tika.metadata.filter.DateNormalizingMetadataFilter"/>
+ <metadataFilter
class="org.apache.tika.metadata.filter.FieldNameMappingFilter">
+ <params>
+ <excludeUnmapped>true</excludeUnmapped>
+ <mappings>
+ <mapping from="X-TIKA:content" to="content_s"/>
+ <mapping from="Content-Length" to="length_i"/>
+ <mapping from="dc:creator" to="creators_ss"/>
+ <mapping from="dc:title" to="title_s"/>
+ <mapping from="Content-Type" to="mime_s"/>
+ <mapping from="X-TIKA:EXCEPTION:container_exception"
to="tika_exception_s"/>
+ </mappings>
+ </params>
+ </metadataFilter>
+ </metadataFilters>
+ <async>
+ <params>
+ <maxForEmitBatchBytes>10000</maxForEmitBatchBytes>
+ <emitMaxEstimatedBytes>100000</emitMaxEstimatedBytes>
+ <emitWithinMillis>10</emitWithinMillis>
+ <numEmitters>1</numEmitters>
+ <numClients>1</numClients>
+ <tikaConfig>{TIKA_CONFIG}</tikaConfig>
+ <forkedJvmArgs>
+ <arg>-Xmx1g</arg>
+ <arg>-XX:ParallelGCThreads=2</arg>
+ <arg>-XX:+ExitOnOutOfMemoryError</arg>
+ <arg>-Dlog4j.configurationFile={LOG4J_PROPERTIES_FILE}</arg>
+ </forkedJvmArgs>
+ <timeoutMillis>60000</timeoutMillis>
+ </params>
+ </async>
+ <fetchers>
+ <fetcher class="org.apache.tika.pipes.fetcher.fs.FileSystemFetcher">
+ <params>
+ <name>fsf</name>
+ <basePath>{PATH_TO_DOCS}</basePath>
+ </params>
+ </fetcher>
+ </fetchers>
+ <emitters>
+ <emitter class="org.apache.tika.pipes.emitter.kafka.KafkaEmitter">
+ <params>
+ <name>ke</name>
+ <topic>{EMITTER_TOPIC}</topic>
+ <bootstrapServers>{BOOTSTRAP_SERVERS}</bootstrapServers>
+ </params>
+ </emitter>
+ <emitter class="org.apache.tika.pipes.emitter.fs.FileSystemEmitter">
+ <params>
+ <name>fse</name>
+ <basePath>/path/to/extracts</basePath>
+ </params>
+ </emitter>
+ </emitters>
+ <pipesIterator
class="org.apache.tika.pipes.pipesiterator.kafka.KafkaPipesIterator">
+ <params>
+ <topic>{PIPE_ITERATOR_TOPIC}</topic>
+ <bootstrapServers>{BOOTSTRAP_SERVERS}</bootstrapServers>
+ <groupId>grpid</groupId>
+ <autoOffsetReset>earliest</autoOffsetReset>
+ <pollDelayMs>1000</pollDelayMs>
+ <fetcherName>fsf</fetcherName>
+ <emitterName>ke</emitterName>
+ </params>
+ </pipesIterator>
+</properties>
\ No newline at end of file
diff --git
a/tika-integration-tests/tika-pipes-solr-integration-tests/src/test/java/org/apache/tika/pipes/solr/tests/TikaPipesSolrTestBase.java
b/tika-integration-tests/tika-pipes-solr-integration-tests/src/test/java/org/apache/tika/pipes/solr/tests/TikaPipesSolrTestBase.java
index 49e51c9ff..c82323bd9 100644
---
a/tika-integration-tests/tika-pipes-solr-integration-tests/src/test/java/org/apache/tika/pipes/solr/tests/TikaPipesSolrTestBase.java
+++
b/tika-integration-tests/tika-pipes-solr-integration-tests/src/test/java/org/apache/tika/pipes/solr/tests/TikaPipesSolrTestBase.java
@@ -30,7 +30,6 @@ import org.apache.http.impl.client.HttpClients;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.impl.LBHttpSolrClient;
-import org.apache.solr.client.solrj.response.UpdateResponse;
import org.apache.solr.common.SolrInputDocument;
import org.apache.tika.cli.TikaCLI;
import org.apache.tika.pipes.HandlerConfig;
diff --git a/tika-parent/pom.xml b/tika-parent/pom.xml
index be096a65f..c17065c89 100644
--- a/tika-parent/pom.xml
+++ b/tika-parent/pom.xml
@@ -389,6 +389,7 @@
<xerces.version>2.12.2</xerces.version>
<xmpcore.version>6.1.11</xmpcore.version>
<zstd.version>1.5.2-3</zstd.version>
+ <kafka.version>3.2.0</kafka.version>
<netty.version>4.1.79.Final</netty.version>
</properties>
diff --git a/tika-pipes/tika-emitters/pom.xml b/tika-pipes/tika-emitters/pom.xml
index 887aad93b..c5777b20b 100644
--- a/tika-pipes/tika-emitters/pom.xml
+++ b/tika-pipes/tika-emitters/pom.xml
@@ -34,6 +34,7 @@
<modules>
<module>tika-emitter-fs</module>
<module>tika-emitter-s3</module>
+ <module>tika-emitter-kafka</module>
<module>tika-emitter-solr</module>
<module>tika-emitter-opensearch</module>
<module>tika-emitter-gcs</module>
diff --git a/tika-pipes/tika-emitters/tika-emitter-kafka/pom.xml
b/tika-pipes/tika-emitters/tika-emitter-kafka/pom.xml
new file mode 100644
index 000000000..24d959503
--- /dev/null
+++ b/tika-pipes/tika-emitters/tika-emitter-kafka/pom.xml
@@ -0,0 +1,132 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
https://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <parent>
+ <artifactId>tika-emitters</artifactId>
+ <groupId>org.apache.tika</groupId>
+ <version>2.4.2-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>tika-emitter-kafka</artifactId>
+ <name>Apache Tika Kafka emitter</name>
+
+
+ <dependencies>
+ <!-- should serialization be provided or bundled? -->
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-serialization</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <archive>
+ <manifestEntries>
+
<Automatic-Module-Name>org.apache.tika.pipes.emitter.kafka</Automatic-Module-Name>
+ </manifestEntries>
+ </archive>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${maven.shade.version}</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <createDependencyReducedPom>
+ false
+ </createDependencyReducedPom>
+ <!-- <filters> -->
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*</exclude>
+ <exclude>LICENSE.txt</exclude>
+ <exclude>NOTICE.txt</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/LICENSE</resource>
+ <file>target/classes/META-INF/LICENSE</file>
+ </transformer>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/NOTICE</resource>
+ <file>target/classes/META-INF/NOTICE</file>
+ </transformer>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/DEPENDENCIES</resource>
+ <file>target/classes/META-INF/DEPENDENCIES</file>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ </plugins>
+ </build>
+
+ <scm>
+ <tag>2.2.1-rc2</tag>
+ </scm>
+</project>
\ No newline at end of file
diff --git
a/tika-pipes/tika-emitters/tika-emitter-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitter.java
b/tika-pipes/tika-emitters/tika-emitter-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitter.java
new file mode 100644
index 000000000..aeee40049
--- /dev/null
+++
b/tika-pipes/tika-emitters/tika-emitter-kafka/src/main/java/org/apache/tika/pipes/emitter/kafka/KafkaEmitter.java
@@ -0,0 +1,284 @@
+/*
+ * 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.apache.tika.config.TikaConfig.mustNotBeEmpty;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+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.common.serialization.StringSerializer;
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.emitter.AbstractEmitter;
+import org.apache.tika.pipes.emitter.TikaEmitterException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Emits the now-parsed documents into a specified Apache Kafka topic.
+ */
+public class KafkaEmitter extends AbstractEmitter implements Initializable {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(KafkaEmitter.class);
+
+ private static final ObjectMapper OM = new ObjectMapper();
+
+ String topic;
+ String bootstrapServers;
+
+ String acks = "all";
+ int lingerMs = 5000;
+ int batchSize = 16384;
+ int bufferMemory = 32 * 1024 * 1024;
+ String compressionType = "none";
+ int connectionsMaxIdleMs = 9 * 60 * 1000;
+ int deliveryTimeoutMs = 120 * 1000;
+ boolean enableIdempotence = false;
+ String interceptorClasses;
+ int maxBlockMs = 60 * 1000;
+ int maxInFlightRequestsPerConnection = 5;
+ int maxRequestSize = 1024 * 1024;
+ int metadataMaxAgeMs = 5 * 60 * 1000;
+ int requestTimeoutMs = 30 * 1000;
+ int retries = Integer.MAX_VALUE;
+ int retryBackoffMs = 100;
+ int transactionTimeoutMs = 60000;
+ String transactionalId;
+ String clientId;
+ String keySerializer;
+ String valueSerializer;
+
+ private Producer<String, String> producer;
+
+ @Field
+ public void setBootstrapServers(String bootstrapServers) {
+ this.bootstrapServers = bootstrapServers;
+ }
+
+ @Field
+ public void setAcks(String acks) {
+ this.acks = acks;
+ }
+
+ @Field
+ public void setLingerMs(int lingerMs) {
+ this.lingerMs = lingerMs;
+ }
+
+ public void setBatchSize(int batchSize) {
+ this.batchSize = batchSize;
+ }
+
+ @Field
+ public void setBufferMemory(int bufferMemory) {
+ this.bufferMemory = bufferMemory;
+ }
+
+ @Field
+ public void setClientId(String clientId) {
+ this.clientId = clientId;
+ }
+
+ @Field
+ public void setCompressionType(String compressionType) {
+ this.compressionType = compressionType;
+ }
+
+ @Field
+ public void setConnectionsMaxIdleMs(int connectionsMaxIdleMs) {
+ this.connectionsMaxIdleMs = connectionsMaxIdleMs;
+ }
+
+ @Field
+ public void setDeliveryTimeoutMs(int deliveryTimeoutMs) {
+ this.deliveryTimeoutMs = deliveryTimeoutMs;
+ }
+
+ @Field
+ public void setEnableIdempotence(boolean enableIdempotence) {
+ this.enableIdempotence = enableIdempotence;
+ }
+
+ @Field
+ public void setInterceptorClasses(String interceptorClasses) {
+ this.interceptorClasses = interceptorClasses;
+ }
+
+ @Field
+ public void setMaxBlockMs(int maxBlockMs) {
+ this.maxBlockMs = maxBlockMs;
+ }
+
+ @Field
+ public void setMaxInFlightRequestsPerConnection(int
maxInFlightRequestsPerConnection) {
+ this.maxInFlightRequestsPerConnection =
maxInFlightRequestsPerConnection;
+ }
+
+ @Field
+ public void setMaxRequestSize(int maxRequestSize) {
+ this.maxRequestSize = maxRequestSize;
+ }
+
+ @Field
+ public void setMetadataMaxAgeMs(int metadataMaxAgeMs) {
+ this.metadataMaxAgeMs = metadataMaxAgeMs;
+ }
+
+ @Field
+ public void setRequestTimeoutMs(int requestTimeoutMs) {
+ this.requestTimeoutMs = requestTimeoutMs;
+ }
+
+ @Field
+ public void setRetries(int retries) {
+ this.retries = retries;
+ }
+
+ @Field
+ public void setRetryBackoffMs(int retryBackoffMs) {
+ this.retryBackoffMs = retryBackoffMs;
+ }
+
+ @Field
+ public void setTransactionTimeoutMs(int transactionTimeoutMs) {
+ this.transactionTimeoutMs = transactionTimeoutMs;
+ }
+
+ @Field
+ public void setTransactionalId(String transactionalId) {
+ this.transactionalId = transactionalId;
+ }
+
+ @Field
+ public void setKeySerializer(String keySerializer) {
+ this.keySerializer = keySerializer;
+ }
+
+ @Field
+ public void setValueSerializer(String valueSerializer) {
+ this.valueSerializer = valueSerializer;
+ }
+
+ @Field
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ @Override
+ public void emit(String emitKey, List<Metadata> metadataList)
+ throws IOException, TikaEmitterException {
+ if (metadataList == null || metadataList.isEmpty()) {
+ throw new TikaEmitterException("metadata list must not be null or
of size 0");
+ }
+ for (Metadata metadata : metadataList) {
+ LOGGER.debug("about to emit to target topic: ({}) path:({})",
topic, emitKey);
+
+ Map<String, Object> fields = new HashMap<>();
+ for (String n : metadata.names()) {
+ String[] vals = metadata.getValues(n);
+ if (vals.length > 1) {
+ LOGGER.warn("Can only write the first value for key {}. I
see {} values.",
+ n,
+ vals.length);
+ }
+ fields.put(n, vals[0]);
+ }
+
+ producer.send(new ProducerRecord<>(topic, emitKey,
OM.writeValueAsString(fields)));
+ }
+ }
+
+ private void safePut(Properties props, String key, Object val) {
+ if (val != null) {
+ props.put(key, val);
+ }
+ }
+
+ @Override
+ public void initialize(Map<String, Param> params) throws
TikaConfigException {
+
+ // create instance for properties to access producer configs
+ Properties props = new Properties();
+
+ //Assign localhost id
+ safePut(props, ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
+
+ //Set acknowledgements for producer requests.
+ safePut(props, ProducerConfig.ACKS_CONFIG, acks);
+
+ //If the request fails, the producer can automatically retry,
+ safePut(props, ProducerConfig.RETRIES_CONFIG, retries);
+
+ //Specify buffer size in config
+ safePut(props, ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
+
+ //Reduce the no of requests less than 0
+ safePut(props, ProducerConfig.LINGER_MS_CONFIG, lingerMs);
+
+ //The buffer.memory controls the total amount of memory available to
the producer for buffering.
+ safePut(props, ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
+
+ safePut(props, ProducerConfig.CLIENT_ID_CONFIG, clientId);
+ safePut(props, ProducerConfig.COMPRESSION_TYPE_CONFIG,
compressionType);
+ safePut(props, ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG,
deliveryTimeoutMs);
+ safePut(props, ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,
enableIdempotence);
+ safePut(props, ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
interceptorClasses);
+ safePut(props, ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
+ safePut(props, ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,
maxInFlightRequestsPerConnection);
+ safePut(props, ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize);
+ safePut(props, ProducerConfig.METADATA_MAX_AGE_CONFIG,
metadataMaxAgeMs);
+ safePut(props, ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,
requestTimeoutMs);
+ safePut(props, ProducerConfig.RETRY_BACKOFF_MS_CONFIG, retryBackoffMs);
+ safePut(props, ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,
transactionTimeoutMs);
+ safePut(props, ProducerConfig.TRANSACTIONAL_ID_CONFIG,
transactionalId);
+
+ safePut(props, ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
serializerClass(keySerializer, StringSerializer.class));
+ safePut(props, ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
serializerClass(valueSerializer, StringSerializer.class));
+
+ producer = new KafkaProducer<>(props);
+ }
+
+ private Object serializerClass(String className, Class defaultClass) {
+ try {
+ return className == null ? defaultClass : Class.forName(className);
+ } catch (ClassNotFoundException e) {
+ LOGGER.error("Could not find key serializer class: {}", className);
+ return null;
+ }
+ }
+
+ @Override
+ public void checkInitialization(InitializableProblemHandler problemHandler)
+ throws TikaConfigException {
+ mustNotBeEmpty("topic", this.topic);
+ mustNotBeEmpty("server", this.bootstrapServers);
+ }
+
+}
diff --git a/tika-pipes/tika-pipes-iterators/pom.xml
b/tika-pipes/tika-pipes-iterators/pom.xml
index 3f3415be8..d5ace7056 100644
--- a/tika-pipes/tika-pipes-iterators/pom.xml
+++ b/tika-pipes/tika-pipes-iterators/pom.xml
@@ -37,6 +37,7 @@
<module>tika-pipes-iterator-csv</module>
<module>tika-pipes-iterator-jdbc</module>
<module>tika-pipes-iterator-s3</module>
+ <module>tika-pipes-iterator-kafka</module>
<module>tika-pipes-iterator-solr</module>
<module>tika-pipes-iterator-gcs</module>
<module>tika-pipes-iterator-az-blob</module>
diff --git a/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/pom.xml
b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/pom.xml
new file mode 100644
index 000000000..086ec23f6
--- /dev/null
+++ b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/pom.xml
@@ -0,0 +1,125 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
https://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <parent>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-pipes-iterators</artifactId>
+ <version>2.4.2-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>tika-pipes-iterator-kafka</artifactId>
+
+ <name>Apache Tika Pipes Iterator - Kafka</name>
+ <url>https://tika.apache.org/</url>
+
+ <dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <version>${log4j2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <configuration>
+ <archive>
+ <manifestEntries>
+
<Automatic-Module-Name>org.apache.tika.pipes.pipesiterator.kafka</Automatic-Module-Name>
+ </manifestEntries>
+ </archive>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>${maven.shade.version}</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <createDependencyReducedPom>
+ false
+ </createDependencyReducedPom>
+ <!-- <filters> -->
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*</exclude>
+ <exclude>LICENSE.txt</exclude>
+ <exclude>NOTICE.txt</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/LICENSE</resource>
+ <file>target/classes/META-INF/LICENSE</file>
+ </transformer>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/NOTICE</resource>
+ <file>target/classes/META-INF/NOTICE</file>
+ </transformer>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
+ <resource>META-INF/DEPENDENCIES</resource>
+ <file>target/classes/META-INF/DEPENDENCIES</file>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ </plugins>
+ </build>
+
+ <scm>
+ <tag>2.2.1-rc2</tag>
+ </scm>
+</project>
diff --git
a/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/main/java/org/apache/tika/pipes/pipesiterator/kafka/KafkaPipesIterator.java
b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/main/java/org/apache/tika/pipes/pipesiterator/kafka/KafkaPipesIterator.java
new file mode 100644
index 000000000..5cb917144
--- /dev/null
+++
b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/main/java/org/apache/tika/pipes/pipesiterator/kafka/KafkaPipesIterator.java
@@ -0,0 +1,174 @@
+/*
+ * 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.pipesiterator.kafka;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.config.TikaConfig;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.FetchEmitTuple;
+import org.apache.tika.pipes.HandlerConfig;
+import org.apache.tika.pipes.emitter.EmitKey;
+import org.apache.tika.pipes.fetcher.FetchKey;
+import org.apache.tika.pipes.pipesiterator.PipesIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class KafkaPipesIterator extends PipesIterator implements Initializable
{
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(KafkaPipesIterator.class);
+ String topic;
+ String bootstrapServers;
+ String keySerializer;
+ String valueSerializer;
+ String groupId;
+ String autoOffsetReset = "earliest";
+ int pollDelayMs = 100;
+ int emitMax = -1;
+ int groupInitialRebalanceDelayMs = 3000;
+
+ private Properties props;
+ private KafkaConsumer<String, String> consumer;
+
+
+ @Field
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ @Field
+ public void setGroupId(String groupId) {
+ this.groupId = groupId;
+ }
+
+ @Field
+ public void setBootstrapServers(String bootstrapServers) {
+ this.bootstrapServers = bootstrapServers;
+ }
+
+ @Field
+ public void setKeySerializer(String keySerializer) {
+ this.keySerializer = keySerializer;
+ }
+
+ @Field
+ public void setAutoOffsetReset(String autoOffsetReset) {
+ this.autoOffsetReset = autoOffsetReset;
+ }
+
+ @Field
+ public void setValueSerializer(String valueSerializer) {
+ this.valueSerializer = valueSerializer;
+ }
+
+ @Field
+ public void setPollDelayMs(int pollDelayMs) {
+ this.pollDelayMs = pollDelayMs;
+ }
+
+ @Field
+ public void setGroupInitialRebalanceDelayMs(int
groupInitialRebalanceDelayMs) {
+ this.groupInitialRebalanceDelayMs = groupInitialRebalanceDelayMs;
+ }
+
+ /**
+ * If the kafka pipe iterator will keep polling for more documents until
it returns an empty result.
+ * If you set emitMax is set to > 0, it will stop polling if the number of
documents you
+ * have emitted so far > emitMax.
+ */
+ @Field
+ public void setEmitMax(int emitMax) {
+ this.emitMax = emitMax;
+ }
+
+ private void safePut(Properties props, String key, Object val) {
+ if (val != null) {
+ props.put(key, val);
+ }
+ }
+
+ @Override
+ public void initialize(Map<String, Param> params) {
+ props = new Properties();
+ safePut(props, ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
+ safePut(props, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
serializerClass(keySerializer, StringDeserializer.class));
+ safePut(props, ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
serializerClass(valueSerializer, StringDeserializer.class));
+ safePut(props, ConsumerConfig.GROUP_ID_CONFIG, groupId);
+ safePut(props, ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
autoOffsetReset);
+ safePut(props, "group.inital.rebalance.delay.ms",
groupInitialRebalanceDelayMs);
+ consumer = new KafkaConsumer<>(props);
+ consumer.subscribe(Arrays.asList(topic));
+ }
+
+ private Object serializerClass(String className, Class defaultClass) {
+ try {
+ return className == null ? defaultClass : Class.forName(className);
+ } catch (ClassNotFoundException e) {
+ LOGGER.error("Could not find key serializer class: {}", className);
+ return null;
+ }
+ }
+
+ @Override
+ public void checkInitialization(InitializableProblemHandler problemHandler)
+ throws TikaConfigException {
+ super.checkInitialization(problemHandler);
+ TikaConfig.mustNotBeEmpty("bootstrapServers", this.bootstrapServers);
+ TikaConfig.mustNotBeEmpty("topic", this.topic);
+ }
+
+ @Override
+ protected void enqueue() throws InterruptedException, TimeoutException {
+ String fetcherName = getFetcherName();
+ String emitterName = getEmitterName();
+ long start = System.currentTimeMillis();
+ int count = 0;
+ HandlerConfig handlerConfig = getHandlerConfig();
+ ConsumerRecords<String, String> records;
+
+ do {
+ records = consumer.poll(Duration.ofMillis(pollDelayMs));
+ for (ConsumerRecord<String, String> r : records) {
+ long elapsed = System.currentTimeMillis() - start;
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("adding ({}) {} in {} ms", count, r.key(),
elapsed);
+ }
+ tryToAdd(new FetchEmitTuple(r.key(), new FetchKey(fetcherName,
+ r.key()),
+ new EmitKey(emitterName, r.key()), new Metadata(),
handlerConfig,
+ getOnParseException()));
+ ++count;
+ }
+ } while ((emitMax > 0 || count < emitMax) && !records.isEmpty());
+ long elapsed = System.currentTimeMillis() - start;
+ LOGGER.info("Finished enqueuing {} files in {} ms", count, elapsed);
+ }
+}
diff --git
a/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/test/java/org/apache/tika/pipes/pipesiterator/kafka/TestKafkaPipesIterator.java
b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/test/java/org/apache/tika/pipes/pipesiterator/kafka/TestKafkaPipesIterator.java
new file mode 100644
index 000000000..152c5101b
--- /dev/null
+++
b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/test/java/org/apache/tika/pipes/pipesiterator/kafka/TestKafkaPipesIterator.java
@@ -0,0 +1,98 @@
+/*
+ * 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.pipesiterator.kafka;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.tika.pipes.FetchEmitTuple;
+import org.apache.tika.pipes.pipesiterator.PipesIterator;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+@Disabled("turn into an actual unit test")
+public class TestKafkaPipesIterator {
+
+ @Test
+ public void testSimple() throws Exception {
+ KafkaPipesIterator it = new KafkaPipesIterator();
+ it.setFetcherName("kafka");
+ it.setGroupId("");//find one
+ it.setBootstrapServers("");//use one
+ it.setTopic("");//select one
+ it.initialize(Collections.EMPTY_MAP);
+ int numConsumers = 6;
+ ArrayBlockingQueue<FetchEmitTuple> queue = new
ArrayBlockingQueue<>(10);
+
+ ExecutorService es = Executors.newFixedThreadPool(numConsumers + 1);
+ ExecutorCompletionService c = new ExecutorCompletionService(es);
+ List<MockFetcher> fetchers = new ArrayList<>();
+ for (int i = 0; i < numConsumers; i++) {
+ MockFetcher fetcher = new MockFetcher(queue);
+ fetchers.add(fetcher);
+ c.submit(fetcher);
+ }
+ for (FetchEmitTuple t : it) {
+ queue.offer(t);
+ }
+ for (int i = 0; i < numConsumers; i++) {
+ queue.offer(PipesIterator.COMPLETED_SEMAPHORE);
+ }
+ int finished = 0;
+ int completed = 0;
+ try {
+ while (finished++ < numConsumers + 1) {
+ Future<Integer> f = c.take();
+ completed += f.get();
+ }
+ } finally {
+ es.shutdownNow();
+ }
+ assertEquals(20000, completed);
+
+ }
+
+ private static class MockFetcher implements Callable<Integer> {
+ private final ArrayBlockingQueue<FetchEmitTuple> queue;
+ private final List<FetchEmitTuple> pairs = new ArrayList<>();
+
+ private MockFetcher(ArrayBlockingQueue<FetchEmitTuple> queue) {
+ this.queue = queue;
+ }
+
+ @Override
+ public Integer call() throws Exception {
+ while (true) {
+ FetchEmitTuple t = queue.poll(1, TimeUnit.HOURS);
+ if (t == PipesIterator.COMPLETED_SEMAPHORE) {
+ return pairs.size();
+ }
+ pairs.add(t);
+ }
+ }
+ }
+}
diff --git
a/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/test/resources/log4j2.xml
b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/test/resources/log4j2.xml
new file mode 100644
index 000000000..c88e66e99
--- /dev/null
+++
b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-kafka/src/test/resources/log4j2.xml
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no" ?>
+
+<!--
+ 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.
+-->
+<Configuration status="WARN">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_ERR">
+ <PatternLayout pattern="%-5p [%t] %d{HH:mm:ss,SSS} %c %m%n"/>
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Root level="info">
+ <AppenderRef ref="Console"/>
+ </Root>
+ </Loggers>
+</Configuration>
\ No newline at end of file
