You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by nd...@apache.org on 2022/10/14 20:30:53 UTC
[tika] branch main updated: TIKA-3879 - add s3 testcontainers integration test (#742)
This is an automated email from the ASF dual-hosted git repository.
ndipiazza 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 527133955 TIKA-3879 - add s3 testcontainers integration test (#742)
527133955 is described below
commit 527133955e65c73ddb52b1e3be624251ab8acd3d
Author: Nicholas DiPiazza <ni...@lucidworks.com>
AuthorDate: Fri Oct 14 15:30:48 2022 -0500
TIKA-3879 - add s3 testcontainers integration test (#742)
* add test containers test, add a few options to make it so you can use minio for s3 endpoint
* remove useless file
* fix a couple sonar qube complaints
* accidentally added another region field instead of using existing one.
* fix import by wildcard
replace assert with Assertion.assertNotNull
* fix import by wildcard
* header
* update checkstyle - got reverted due to new intellij profile
* accidentally wasn't using hamcrest version
* fix checkstyle.
* fix checkstyle in other 2 files now.
* add intellij-code-style.xml so people can import it
fix issue with hamcrest - use the testcontainers version to avoid needing to add 1.x on the classpath.
* fix a few more checkstyle things
Co-authored-by: Nicholas DiPiazza <ni...@apache.org>
---
.../tika-pipes-s3-integration-tests/pom.xml | 23 +++
.../tika/pipes/s3/tests/S3PipeIntegrationTest.java | 162 +++++++++++++++++++++
.../src/test/resources/docker-compose.yml | 10 ++
.../src/test/resources/log4j2.xml | 32 ++++
.../resources/tika-config-s3-integration-test.xml | 125 ++++++++++++++++
tika-parent/intellij-code-style.xml | 55 +++++++
.../apache/tika/pipes/emitter/s3/S3Emitter.java | 54 ++++++-
.../apache/tika/pipes/fetcher/s3/S3Fetcher.java | 66 +++++++--
.../pipes/pipesiterator/s3/S3PipesIterator.java | 77 ++++++++--
9 files changed, 571 insertions(+), 33 deletions(-)
diff --git a/tika-integration-tests/tika-pipes-s3-integration-tests/pom.xml b/tika-integration-tests/tika-pipes-s3-integration-tests/pom.xml
index e14c81b92..414619fa6 100644
--- a/tika-integration-tests/tika-pipes-s3-integration-tests/pom.xml
+++ b/tika-integration-tests/tika-pipes-s3-integration-tests/pom.xml
@@ -53,6 +53,29 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <version>${test.containers.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-core</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>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j2-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<scm>
diff --git a/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/java/org/apache/tika/pipes/s3/tests/S3PipeIntegrationTest.java b/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/java/org/apache/tika/pipes/s3/tests/S3PipeIntegrationTest.java
new file mode 100644
index 000000000..0cd6d0d78
--- /dev/null
+++ b/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/java/org/apache/tika/pipes/s3/tests/S3PipeIntegrationTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.s3.tests;
+
+import java.io.File;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.HashSet;
+import java.util.Set;
+
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.S3Object;
+import org.apache.commons.io.IOUtils;
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.DockerComposeContainer;
+import org.testcontainers.shaded.org.apache.commons.io.FileUtils;
+import org.testcontainers.shaded.org.hamcrest.MatcherAssert;
+import org.testcontainers.shaded.org.hamcrest.Matchers;
+
+import org.apache.tika.cli.TikaCLI;
+import org.apache.tika.pipes.HandlerConfig;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class S3PipeIntegrationTest {
+ private static final Logger LOG = LoggerFactory.getLogger(S3PipeIntegrationTest.class);
+
+ public static final int MAX_STARTUP_TIMEOUT = 120;
+ private static final DockerComposeContainer<?> minioContainer = new DockerComposeContainer<>(
+ new File("src/test/resources/docker-compose.yml")).withStartupTimeout(
+ Duration.of(MAX_STARTUP_TIMEOUT, ChronoUnit.SECONDS))
+ .withExposedService("minio-service", 9000);
+ private static final String MINIO_ENDPOINT = "http://localhost:9000";
+ private static final String ACCESS_KEY = "minio";
+ private static final String SECRET_KEY = "minio123";
+ private static final String FETCH_BUCKET = "fetch-bucket";
+ private static final String EMIT_BUCKET = "emit-bucket";
+
+ private static final String REGION = Regions.US_EAST_1.getName();
+
+ private AmazonS3 s3Client;
+
+ private final File testFileFolder = new File("target", "test-files");
+ private final Set<String> testFiles = new HashSet<>();
+
+ private void createTestFiles() {
+ if (testFileFolder.mkdirs()) {
+ LOG.info("Created test folder: {}", testFileFolder);
+ }
+ int numDocs = 42;
+ for (int i = 0; i < numDocs; ++i) {
+ String nextFileName = "test-" + i + ".html";
+ testFiles.add(nextFileName);
+ s3Client.putObject(FETCH_BUCKET, nextFileName,
+ "<html><body>body-of-" + nextFileName + "</body></html>");
+ }
+ }
+
+ @BeforeAll
+ void setupMinio() {
+ minioContainer.start();
+ initializeS3Client();
+ }
+
+ @AfterAll
+ void closeMinio() {
+ minioContainer.close();
+ }
+
+ private void initializeS3Client() {
+ AwsClientBuilder.EndpointConfiguration endpoint =
+ new AwsClientBuilder.EndpointConfiguration(MINIO_ENDPOINT, REGION);
+ s3Client = AmazonS3ClientBuilder.standard().withCredentials(
+ new AWSStaticCredentialsProvider(new BasicAWSCredentials(ACCESS_KEY, SECRET_KEY)))
+ .withEndpointConfiguration(endpoint).withPathStyleAccessEnabled(true).build();
+ }
+
+ @Test
+ void s3PipelineIteratorS3FetcherAndS3Emitter() throws Exception {
+
+ // create s3 bucket for fetches and for emits
+ s3Client.createBucket(FETCH_BUCKET);
+ s3Client.createBucket(EMIT_BUCKET);
+
+ // create some test files and insert into fetch bucket
+ createTestFiles();
+
+ // Let's fetch it
+ 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")) {
+ Assertions.assertNotNull(is);
+ FileUtils.copyInputStreamToFile(is, log4jPropFile);
+ }
+ String tikaConfigTemplateXml;
+ try (InputStream is = this.getClass()
+ .getResourceAsStream("/tika-config-s3-integration-test.xml")) {
+ assert is != null;
+ tikaConfigTemplateXml = IOUtils.toString(is, StandardCharsets.UTF_8);
+ }
+ 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);
+ }
+
+ for (String testFile : testFiles) {
+ S3Object object = s3Client.getObject(EMIT_BUCKET, testFile + ".json");
+ Assertions.assertNotNull(object);
+ String data = IOUtils.toString(object.getObjectContent(), StandardCharsets.UTF_8);
+ MatcherAssert.assertThat(
+ "Should be able to read the parsed body of the HTML file as the body of the document",
+ data, Matchers.containsString("body-of-" + testFile));
+ }
+ }
+
+ @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_BUCKET}", FETCH_BUCKET)
+ .replace("{EMIT_BUCKET}", EMIT_BUCKET).replace("{FETCH_BUCKET}", FETCH_BUCKET)
+ .replace("{ACCESS_KEY}", ACCESS_KEY).replace("{SECRET_KEY}", SECRET_KEY)
+ .replace("{ENDPOINT_CONFIGURATION_SERVICE}", MINIO_ENDPOINT)
+ .replace("{REGION}", REGION);
+ }
+}
diff --git a/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/docker-compose.yml b/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/docker-compose.yml
new file mode 100644
index 000000000..02a6ac5a4
--- /dev/null
+++ b/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/docker-compose.yml
@@ -0,0 +1,10 @@
+version: '3.7'
+services:
+ minio-service:
+ image: quay.io/minio/minio
+ command: minio server /data
+ ports:
+ - "9000:9000"
+ environment:
+ MINIO_ROOT_USER: minio
+ MINIO_ROOT_PASSWORD: minio123
\ No newline at end of file
diff --git a/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/log4j2.xml b/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/log4j2.xml
new file mode 100644
index 000000000..c88e66e99
--- /dev/null
+++ b/tika-integration-tests/tika-pipes-s3-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-s3-integration-tests/src/test/resources/tika-config-s3-integration-test.xml b/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/tika-config-s3-integration-test.xml
new file mode 100644
index 000000000..891e2cf19
--- /dev/null
+++ b/tika-integration-tests/tika-pipes-s3-integration-tests/src/test/resources/tika-config-s3-integration-test.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.
+-->
+<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.s3.S3Fetcher">
+ <params>
+ <name>s3f</name>
+ <region>{REGION}</region>
+ <bucket>{FETCH_BUCKET}</bucket>
+ <credentialsProvider>key_secret</credentialsProvider>
+ <accessKey>{ACCESS_KEY}</accessKey>
+ <secretKey>{SECRET_KEY}</secretKey>
+ <endpointConfigurationService>{ENDPOINT_CONFIGURATION_SERVICE}</endpointConfigurationService>
+ <pathStyleAccessEnabled>true</pathStyleAccessEnabled>
+ </params>
+ </fetcher>
+ </fetchers>
+ <pipesIterator class="org.apache.tika.pipes.pipesiterator.s3.S3PipesIterator">
+ <params>
+ <emitterName>s3e</emitterName>
+ <fetcherName>s3f</fetcherName>
+ <region>{REGION}</region>
+ <bucket>{PIPE_ITERATOR_BUCKET}</bucket>
+ <credentialsProvider>key_secret</credentialsProvider>
+ <accessKey>{ACCESS_KEY}</accessKey>
+ <secretKey>{SECRET_KEY}</secretKey>
+ <endpointConfigurationService>{ENDPOINT_CONFIGURATION_SERVICE}</endpointConfigurationService>
+ <pathStyleAccessEnabled>true</pathStyleAccessEnabled>
+ </params>
+ </pipesIterator>
+ <emitters>
+ <emitter class="org.apache.tika.pipes.emitter.s3.S3Emitter">
+ <params>
+ <name>s3e</name>
+ <region>{REGION}</region>
+ <bucket>{EMIT_BUCKET}</bucket>
+ <credentialsProvider>key_secret</credentialsProvider>
+ <accessKey>{ACCESS_KEY}</accessKey>
+ <secretKey>{SECRET_KEY}</secretKey>
+ <endpointConfigurationService>{ENDPOINT_CONFIGURATION_SERVICE}</endpointConfigurationService>
+ <pathStyleAccessEnabled>true</pathStyleAccessEnabled>
+ </params>
+ </emitter>
+ </emitters>
+</properties>
\ No newline at end of file
diff --git a/tika-parent/intellij-code-style.xml b/tika-parent/intellij-code-style.xml
new file mode 100644
index 000000000..efeee3792
--- /dev/null
+++ b/tika-parent/intellij-code-style.xml
@@ -0,0 +1,55 @@
+<code_scheme name="Default" version="173">
+ <JavaCodeStyleSettings>
+ <option name="ANNOTATION_PARAMETER_WRAP" value="1" />
+ <option name="CLASS_COUNT_TO_USE_IMPORT_ON_DEMAND" value="100000" />
+ <option name="NAMES_COUNT_TO_USE_IMPORT_ON_DEMAND" value="10000" />
+ <option name="PACKAGES_TO_USE_IMPORT_ON_DEMAND">
+ <value />
+ </option>
+ <option name="IMPORT_LAYOUT_TABLE">
+ <value>
+ <package name="" withSubpackages="true" static="true" />
+ <emptyLine />
+ <package name="java" withSubpackages="true" static="false" />
+ <package name="javax" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="" withSubpackages="true" static="false" />
+ <emptyLine />
+ <package name="org.apache.tika" withSubpackages="true" static="false" />
+ </value>
+ </option>
+ </JavaCodeStyleSettings>
+ <editorconfig>
+ <option name="ENABLED" value="false" />
+ </editorconfig>
+ <codeStyleSettings language="JAVA">
+ <option name="RIGHT_MARGIN" value="100" />
+ <option name="KEEP_LINE_BREAKS" value="false" />
+ <option name="CALL_PARAMETERS_WRAP" value="1" />
+ <option name="METHOD_PARAMETERS_WRAP" value="1" />
+ <option name="RESOURCE_LIST_WRAP" value="1" />
+ <option name="EXTENDS_LIST_WRAP" value="1" />
+ <option name="THROWS_LIST_WRAP" value="1" />
+ <option name="EXTENDS_KEYWORD_WRAP" value="1" />
+ <option name="THROWS_KEYWORD_WRAP" value="1" />
+ <option name="METHOD_CALL_CHAIN_WRAP" value="1" />
+ <option name="BINARY_OPERATION_WRAP" value="1" />
+ <option name="TERNARY_OPERATION_WRAP" value="1" />
+ <option name="FOR_STATEMENT_WRAP" value="1" />
+ <option name="ARRAY_INITIALIZER_WRAP" value="1" />
+ <option name="ASSIGNMENT_WRAP" value="1" />
+ <option name="ASSERT_STATEMENT_WRAP" value="1" />
+ <option name="IF_BRACE_FORCE" value="3" />
+ <option name="PARAMETER_ANNOTATION_WRAP" value="1" />
+ <option name="VARIABLE_ANNOTATION_WRAP" value="1" />
+ <option name="ENUM_CONSTANTS_WRAP" value="1" />
+ <option name="WRAP_ON_TYPING" value="1" />
+ </codeStyleSettings>
+ <codeStyleSettings language="XML">
+ <indentOptions>
+ <option name="INDENT_SIZE" value="2" />
+ <option name="CONTINUATION_INDENT_SIZE" value="4" />
+ <option name="TAB_SIZE" value="2" />
+ </indentOptions>
+ </codeStyleSettings>
+</code_scheme>
\ No newline at end of file
diff --git a/tika-pipes/tika-emitters/tika-emitter-s3/src/main/java/org/apache/tika/pipes/emitter/s3/S3Emitter.java b/tika-pipes/tika-emitters/tika-emitter-s3/src/main/java/org/apache/tika/pipes/emitter/s3/S3Emitter.java
index 9190c5880..d03c3c9be 100644
--- a/tika-pipes/tika-emitters/tika-emitter-s3/src/main/java/org/apache/tika/pipes/emitter/s3/S3Emitter.java
+++ b/tika-pipes/tika-emitters/tika-emitter-s3/src/main/java/org/apache/tika/pipes/emitter/s3/S3Emitter.java
@@ -33,8 +33,11 @@ import java.util.Map;
import com.amazonaws.AmazonClientException;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
+import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.ObjectMetadata;
@@ -99,10 +102,14 @@ public class S3Emitter extends AbstractEmitter implements Initializable, StreamE
private String profile;
private String bucket;
private String credentialsProvider;
+ private String accessKey;
+ private String secretKey;
+ private String endpointConfigurationService;
private String fileExtension = "json";
private boolean spoolToTemp = true;
private String prefix = null;
private int maxConnections = ClientConfiguration.DEFAULT_MAX_CONNECTIONS;
+ private boolean pathStyleAccessEnabled = false;
private AmazonS3 s3Client;
/**
@@ -238,9 +245,10 @@ public class S3Emitter extends AbstractEmitter implements Initializable, StreamE
@Field
public void setCredentialsProvider(String credentialsProvider) {
- if (!credentialsProvider.equals("profile") && !credentialsProvider.equals("instance")) {
+ if (!credentialsProvider.equals("profile") && !credentialsProvider.equals("instance")
+ && !credentialsProvider.equals("key_secret")) {
throw new IllegalArgumentException(
- "credentialsProvider must be either 'profile' or instance'");
+ "credentialsProvider must be either 'profile', 'instance' or 'key_secret'");
}
this.credentialsProvider = credentialsProvider;
}
@@ -256,6 +264,16 @@ public class S3Emitter extends AbstractEmitter implements Initializable, StreamE
this.fileExtension = fileExtension;
}
+ @Field
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ @Field
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
/**
* maximum number of http connections allowed. This should be
* greater than or equal to the number of threads emitting to S3.
@@ -266,6 +284,12 @@ public class S3Emitter extends AbstractEmitter implements Initializable, StreamE
public void setMaxConnections(int maxConnections) {
this.maxConnections = maxConnections;
}
+
+ @Field
+ public void setEndpointConfigurationService(String endpointConfigurationService) {
+ this.endpointConfigurationService = endpointConfigurationService;
+ }
+
/**
* This initializes the s3 client. Note, we wrap S3's RuntimeExceptions,
* e.g. AmazonClientException in a TikaConfigException.
@@ -276,22 +300,32 @@ public class S3Emitter extends AbstractEmitter implements Initializable, StreamE
@Override
public void initialize(Map<String, Param> params) throws TikaConfigException {
//params have already been set...ignore them
- AWSCredentialsProvider provider = null;
+ AWSCredentialsProvider provider;
if ("instance".equals(credentialsProvider)) {
provider = InstanceProfileCredentialsProvider.getInstance();
} else if ("profile".equals(credentialsProvider)) {
provider = new ProfileCredentialsProvider(profile);
+ } else if (credentialsProvider.equals("key_secret")) {
+ provider = new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey));
} else {
throw new TikaConfigException("credentialsProvider must be set and " +
- "must be either 'instance' or 'profile'");
+ "must be either 'instance', 'profile' or 'key_secret'");
}
ClientConfiguration clientConfig = new ClientConfiguration()
.withMaxConnections(maxConnections);
try {
- s3Client =
- AmazonS3ClientBuilder.standard()
- .withClientConfiguration(clientConfig).withRegion(region).withCredentials(provider)
- .build();
+ AmazonS3ClientBuilder amazonS3ClientBuilder = AmazonS3ClientBuilder.standard()
+ .withClientConfiguration(clientConfig)
+ .withCredentials(provider)
+ .withPathStyleAccessEnabled(pathStyleAccessEnabled);
+ if (!StringUtils.isBlank(endpointConfigurationService)) {
+ amazonS3ClientBuilder.setEndpointConfiguration(
+ new AwsClientBuilder
+ .EndpointConfiguration(endpointConfigurationService, region));
+ } else {
+ amazonS3ClientBuilder.withRegion(region);
+ }
+ s3Client = amazonS3ClientBuilder.build();
} catch (AmazonClientException e) {
throw new TikaConfigException("can't initialize s3 emitter", e);
}
@@ -304,4 +338,8 @@ public class S3Emitter extends AbstractEmitter implements Initializable, StreamE
mustNotBeEmpty("region", this.region);
}
+ @Field
+ public void setPathStyleAccessEnabled(boolean pathStyleAccessEnabled) {
+ this.pathStyleAccessEnabled = pathStyleAccessEnabled;
+ }
}
diff --git a/tika-pipes/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java
index 9fb4a23e8..95760e145 100644
--- a/tika-pipes/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java
+++ b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java
@@ -29,8 +29,11 @@ import java.util.Map;
import com.amazonaws.AmazonClientException;
import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
+import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.GetObjectRequest;
@@ -65,6 +68,9 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
private String region;
private String bucket;
private String profile;
+ private String accessKey;
+ private String secretKey;
+ private String endpointConfigurationService;
private String prefix;
private String credentialsProvider;
private boolean extractUserMetadata = true;
@@ -72,9 +78,9 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
private AmazonS3 s3Client;
private boolean spoolToTemp = true;
private int retries = 0;
-
private long sleepBeforeRetryMillis = 30000;
private long maxLength = -1;
+ private boolean pathStyleAccessEnabled = false;
@Override
public InputStream fetch(String fetchKey, Metadata metadata) throws TikaException, IOException {
@@ -115,7 +121,7 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
long elapsed = System.currentTimeMillis() - start;
LOGGER.debug("total to fetch {}", elapsed);
return is;
- } catch (AmazonClientException e ) {
+ } catch (AmazonClientException e) {
//TODO -- filter exceptions -- if the file doesn't exist, don't retry
LOGGER.warn("client exception fetching on retry=" + tries, e);
ex = new IOException(e);
@@ -201,16 +207,18 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
/**
* prefix to prepend to the fetch key before fetching.
* This will automatically add a '/' at the end.
+ *
* @param prefix
*/
@Field
public void setPrefix(String prefix) {
//guarantee that the prefix ends with /
- if (! prefix.endsWith("/")) {
+ if (!prefix.endsWith("/")) {
prefix += "/";
}
this.prefix = prefix;
}
+
/**
* Whether or not to extract user metadata from the S3Object
*
@@ -233,9 +241,10 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
@Field
public void setCredentialsProvider(String credentialsProvider) {
- if (!credentialsProvider.equals("profile") && !credentialsProvider.equals("instance")) {
+ if (!credentialsProvider.equals("profile") && !credentialsProvider.equals("instance")
+ && !credentialsProvider.equals("key_secret")) {
throw new IllegalArgumentException(
- "credentialsProvider must be either 'profile' or 'instance'");
+ "credentialsProvider must be either 'profile', 'instance' or 'key_secret'");
}
this.credentialsProvider = credentialsProvider;
}
@@ -250,6 +259,16 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
this.sleepBeforeRetryMillis = sleepBeforeRetryMillis;
}
+ @Field
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ @Field
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
/**
* This initializes the s3 client. Note, we wrap S3's RuntimeExceptions,
* e.g. AmazonClientException in a TikaConfigException.
@@ -260,23 +279,34 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
@Override
public void initialize(Map<String, Param> params) throws TikaConfigException {
//params have already been set...ignore them
- AWSCredentialsProvider provider = null;
- if ("instance".equals(credentialsProvider)) {
+ AWSCredentialsProvider provider;
+ if (credentialsProvider.equals("instance")) {
provider = InstanceProfileCredentialsProvider.getInstance();
- } else if ("profile".equals(credentialsProvider)) {
+ } else if (credentialsProvider.equals("profile")) {
provider = new ProfileCredentialsProvider(profile);
+ } else if (credentialsProvider.equals("key_secret")) {
+ provider = new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey));
} else {
throw new TikaConfigException("credentialsProvider must be set and " +
- "must be either 'instance' or 'profile'");
+ "must be either 'instance', 'profile' or 'key_secret'");
}
ClientConfiguration clientConfiguration = new ClientConfiguration()
- .withMaxConnections(maxConnections);
+ .withMaxConnections(maxConnections);
try {
synchronized (clientLock) {
- s3Client = AmazonS3ClientBuilder.standard()
+ AmazonS3ClientBuilder amazonS3ClientBuilder = AmazonS3ClientBuilder.standard()
.withClientConfiguration(clientConfiguration)
- .withRegion(region)
- .withCredentials(provider).build();
+ .withPathStyleAccessEnabled(pathStyleAccessEnabled)
+ .withCredentials(provider);
+ if (!StringUtils.isBlank(endpointConfigurationService)) {
+ amazonS3ClientBuilder.setEndpointConfiguration(
+ new AwsClientBuilder
+ .EndpointConfiguration(endpointConfigurationService, region));
+ } else {
+ amazonS3ClientBuilder.withRegion(region);
+ }
+ s3Client = amazonS3ClientBuilder.build();
+
}
} catch (AmazonClientException e) {
throw new TikaConfigException("can't initialize s3 fetcher", e);
@@ -289,4 +319,14 @@ public class S3Fetcher extends AbstractFetcher implements Initializable, RangeFe
mustNotBeEmpty("bucket", this.bucket);
mustNotBeEmpty("region", this.region);
}
+
+ @Field
+ public void setEndpointConfigurationService(String endpointConfigurationService) {
+ this.endpointConfigurationService = endpointConfigurationService;
+ }
+
+ @Field
+ public void setPathStyleAccessEnabled(boolean pathStyleAccessEnabled) {
+ this.pathStyleAccessEnabled = pathStyleAccessEnabled;
+ }
}
diff --git a/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-s3/src/main/java/org/apache/tika/pipes/pipesiterator/s3/S3PipesIterator.java b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-s3/src/main/java/org/apache/tika/pipes/pipesiterator/s3/S3PipesIterator.java
index 3a7005bfd..eee1b8173 100644
--- a/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-s3/src/main/java/org/apache/tika/pipes/pipesiterator/s3/S3PipesIterator.java
+++ b/tika-pipes/tika-pipes-iterators/tika-pipes-iterator-s3/src/main/java/org/apache/tika/pipes/pipesiterator/s3/S3PipesIterator.java
@@ -25,9 +25,13 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.amazonaws.AmazonClientException;
+import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
+import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.iterable.S3Objects;
@@ -47,18 +51,30 @@ 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.apache.tika.utils.StringUtils;
public class S3PipesIterator extends PipesIterator implements Initializable {
private static final Logger LOGGER = LoggerFactory.getLogger(S3PipesIterator.class);
private String prefix = "";
private String region;
+ private String accessKey;
+ private String secretKey;
+ private String endpointConfigurationService;
private String credentialsProvider;
private String profile;
private String bucket;
private Pattern fileNamePattern = null;
+ private int maxConnections = ClientConfiguration.DEFAULT_MAX_CONNECTIONS;
+ private boolean pathStyleAccessEnabled = false;
+
private AmazonS3 s3Client;
+ @Field
+ public void setEndpointConfigurationService(String endpointConfigurationService) {
+ this.endpointConfigurationService = endpointConfigurationService;
+ }
+
@Field
public void setBucket(String bucket) {
this.bucket = bucket;
@@ -79,11 +95,27 @@ public class S3PipesIterator extends PipesIterator implements Initializable {
this.prefix = prefix;
}
+ @Field
+ public void setAccessKey(String accessKey) {
+ this.accessKey = accessKey;
+ }
+
+ @Field
+ public void setMaxConnections(int maxConnections) {
+ this.maxConnections = maxConnections;
+ }
+
+ @Field
+ public void setSecretKey(String secretKey) {
+ this.secretKey = secretKey;
+ }
+
@Field
public void setCredentialsProvider(String credentialsProvider) {
- if (!credentialsProvider.equals("profile") && !credentialsProvider.equals("instance")) {
+ if (!credentialsProvider.equals("profile") && !credentialsProvider.equals("instance")
+ && !credentialsProvider.equals("key_secret")) {
throw new IllegalArgumentException(
- "credentialsProvider must be either 'profile' or instance'");
+ "credentialsProvider must be either 'profile', 'instance' or 'key_secret'");
}
this.credentialsProvider = credentialsProvider;
}
@@ -93,6 +125,16 @@ public class S3PipesIterator extends PipesIterator implements Initializable {
this.fileNamePattern = Pattern.compile(fileNamePattern);
}
+ @Field
+ public void setFileNamePattern(Pattern fileNamePattern) {
+ this.fileNamePattern = fileNamePattern;
+ }
+
+ @Field
+ public void setPathStyleAccessEnabled(boolean pathStyleAccessEnabled) {
+ this.pathStyleAccessEnabled = pathStyleAccessEnabled;
+ }
+
/**
* This initializes the s3 client. Note, we wrap S3's RuntimeExceptions,
* e.g. AmazonClientException in a TikaConfigException.
@@ -103,19 +145,33 @@ public class S3PipesIterator extends PipesIterator implements Initializable {
@Override
public void initialize(Map<String, Param> params) throws TikaConfigException {
//params have already been set...ignore them
- AWSCredentialsProvider provider = null;
- if ("instance".equals(credentialsProvider)) {
+ AWSCredentialsProvider provider;
+ if (credentialsProvider.equals("instance")) {
provider = InstanceProfileCredentialsProvider.getInstance();
- } else if ("profile".equals(credentialsProvider)) {
+ } else if (credentialsProvider.equals("profile")) {
provider = new ProfileCredentialsProvider(profile);
+ } else if (credentialsProvider.equals("key_secret")) {
+ provider = new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey));
} else {
throw new TikaConfigException("credentialsProvider must be set and " +
- "must be either 'instance' or 'profile'");
+ "must be either 'instance', 'profile' or 'key_secret'");
}
+ ClientConfiguration clientConfig = new ClientConfiguration()
+ .withMaxConnections(maxConnections);
try {
- s3Client = AmazonS3ClientBuilder.standard().withRegion(region).withCredentials(provider)
- .build();
+ AmazonS3ClientBuilder amazonS3ClientBuilder = AmazonS3ClientBuilder.standard()
+ .withClientConfiguration(clientConfig)
+ .withCredentials(provider)
+ .withPathStyleAccessEnabled(pathStyleAccessEnabled);
+ if (!StringUtils.isBlank(endpointConfigurationService)) {
+ amazonS3ClientBuilder.setEndpointConfiguration(
+ new AwsClientBuilder
+ .EndpointConfiguration(endpointConfigurationService, region));
+ } else {
+ amazonS3ClientBuilder.withRegion(region);
+ }
+ s3Client = amazonS3ClientBuilder.build();
} catch (AmazonClientException e) {
throw new TikaConfigException("can't initialize s3 pipesiterator", e);
}
@@ -159,9 +215,6 @@ public class S3PipesIterator extends PipesIterator implements Initializable {
private boolean accept(Matcher fileNameMatcher, String key) {
String fName = FilenameUtils.getName(key);
- if (fileNameMatcher.reset(fName).find()) {
- return true;
- }
- return false;
+ return fileNameMatcher.reset(fName).find();
}
}