Skip to content

Commit d21a66f

Browse files
THausherrTHausherr
committed
TIKA-4525: migrate to aws v2
1 parent 879d1a3 commit d21a66f

2 files changed

Lines changed: 69 additions & 49 deletions

File tree

  • tika-pipes/tika-emitters/tika-emitter-s3

‎tika-pipes/tika-emitters/tika-emitter-s3/pom.xml‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,12 @@
3737
<scope>provided</scope>
3838
</dependency>
3939
<dependency>
40-
<groupId>com.amazonaws</groupId>
41-
<artifactId>aws-java-sdk-s3</artifactId>
40+
<groupId>software.amazon.awssdk</groupId>
41+
<artifactId>s3</artifactId>
42+
</dependency>
43+
<dependency>
44+
<groupId>software.amazon.awssdk</groupId>
45+
<artifactId>apache-client</artifactId>
4246
</dependency>
4347
<dependency>
4448
<groupId>org.apache.logging.log4j</groupId>

‎tika-pipes/tika-emitters/tika-emitter-s3/src/main/java/org/apache/tika/pipes/emitter/s3/S3Emitter.java‎

Lines changed: 63 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -17,41 +17,44 @@
1717
package org.apache.tika.pipes.emitter.s3;
1818

1919
import static org.apache.tika.config.TikaConfig.mustNotBeEmpty;
20+
import static software.amazon.awssdk.http.SdkHttpConfigurationOption.MAX_CONNECTIONS;
2021

2122
import java.io.BufferedWriter;
2223
import java.io.IOException;
2324
import java.io.InputStream;
2425
import java.io.OutputStreamWriter;
2526
import java.io.Writer;
27+
import java.net.URI;
28+
import java.net.URISyntaxException;
2629
import java.nio.charset.StandardCharsets;
2730
import java.nio.file.Files;
2831
import java.nio.file.Path;
2932
import java.nio.file.StandardOpenOption;
33+
import java.util.HashMap;
3034
import java.util.List;
3135
import java.util.Map;
3236

33-
import com.amazonaws.AmazonClientException;
34-
import com.amazonaws.ClientConfiguration;
35-
import com.amazonaws.auth.AWSCredentialsProvider;
36-
import com.amazonaws.auth.AWSStaticCredentialsProvider;
37-
import com.amazonaws.auth.BasicAWSCredentials;
38-
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
39-
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
40-
import com.amazonaws.client.builder.AwsClientBuilder;
41-
import com.amazonaws.services.s3.AmazonS3;
42-
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
43-
import com.amazonaws.services.s3.model.ObjectMetadata;
44-
import com.amazonaws.services.s3.model.PutObjectRequest;
4537
import org.apache.commons.io.output.UnsynchronizedByteArrayOutputStream;
4638
import org.slf4j.Logger;
4739
import org.slf4j.LoggerFactory;
40+
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
41+
import software.amazon.awssdk.core.checksums.RequestChecksumCalculation;
42+
import software.amazon.awssdk.core.sync.RequestBody;
43+
import software.amazon.awssdk.http.SdkHttpClient;
44+
import software.amazon.awssdk.http.SdkHttpConfigurationOption;
45+
import software.amazon.awssdk.http.apache.ApacheHttpClient;
46+
import software.amazon.awssdk.regions.Region;
47+
import software.amazon.awssdk.services.s3.S3Client;
48+
import software.amazon.awssdk.services.s3.S3ClientBuilder;
49+
import software.amazon.awssdk.services.s3.S3Configuration;
50+
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
51+
import software.amazon.awssdk.services.s3.model.S3Exception;
4852

4953
import org.apache.tika.config.Field;
5054
import org.apache.tika.config.Initializable;
5155
import org.apache.tika.config.InitializableProblemHandler;
5256
import org.apache.tika.config.Param;
5357
import org.apache.tika.exception.TikaConfigException;
54-
import org.apache.tika.exception.TikaException;
5558
import org.apache.tika.io.TemporaryResources;
5659
import org.apache.tika.io.TikaInputStream;
5760
import org.apache.tika.metadata.Metadata;
@@ -109,20 +112,22 @@ public class S3Emitter extends AbstractEmitter implements Initializable, StreamE
109112
private String fileExtension = "json";
110113
private boolean spoolToTemp = true;
111114
private String prefix = null;
112-
private int maxConnections = ClientConfiguration.DEFAULT_MAX_CONNECTIONS;
115+
private int maxConnections = SdkHttpConfigurationOption.GLOBAL_HTTP_DEFAULTS.get(MAX_CONNECTIONS);
113116
private boolean pathStyleAccessEnabled = false;
114-
private AmazonS3 s3Client;
117+
private S3Client s3Client;
115118

116119
/**
117120
* Requires the src-bucket/path/to/my/file.txt in the {@link TikaCoreProperties#SOURCE_PATH}.
118121
*
122+
* @param emitKey
119123
* @param metadataList
124+
* @param parseContext
120125
* @throws IOException
121-
* @throws TikaException
126+
* @throws TikaEmitterException
122127
*/
123128
@Override
124129
public void emit(String emitKey, List<Metadata> metadataList, ParseContext parseContext) throws IOException, TikaEmitterException {
125-
if (metadataList == null || metadataList.size() == 0) {
130+
if (metadataList == null || metadataList.isEmpty()) {
126131
throw new TikaEmitterException("metadata list must not be null or of size 0");
127132
}
128133

@@ -158,7 +163,9 @@ public void emit(String emitKey, List<Metadata> metadataList, ParseContext parse
158163
* @param path -- object path, not including the bucket
159164
* @param is inputStream to copy
160165
* @param userMetadata this will be written to the s3 ObjectMetadata's userMetadata
161-
* @throws TikaEmitterException or IOexception if there is a Runtime s3 client exception
166+
* @param parseContext
167+
* @throws IOException if there is a Runtime s3 client exception
168+
* @throws TikaEmitterException if there is a Runtime s3 client exception
162169
*/
163170
@Override
164171
public void emit(String path, InputStream is, Metadata userMetadata, ParseContext parseContext) throws IOException, TikaEmitterException {
@@ -173,31 +180,34 @@ public void emit(String path, InputStream is, Metadata userMetadata, ParseContex
173180

174181
LOGGER.debug("about to emit to target bucket: ({}) path:({})", bucket, path);
175182

176-
ObjectMetadata objectMetadata = new ObjectMetadata();
183+
Map<String,String> metadataMap = new HashMap<>();
177184
for (String n : userMetadata.names()) {
178185
String[] vals = userMetadata.getValues(n);
179186
if (vals.length > 1) {
180187
LOGGER.warn("Can only write the first value for key {}. I see {} values.", n, vals.length);
181188
}
182-
objectMetadata.addUserMetadata(n, vals[0]);
189+
metadataMap.put(n, vals[0]);
183190
}
184191
//In practice, sending a file is more robust
185192
//We ran into stream reset issues during digesting, and aws doesn't
186193
//like putObjects for streams without lengths
187194
if (is instanceof TikaInputStream) {
188195
if (((TikaInputStream) is).hasFile()) {
189196
try {
190-
PutObjectRequest putObjectRequest = new PutObjectRequest(bucket, path, ((TikaInputStream) is).getFile()).withMetadata(objectMetadata);
191-
s3Client.putObject(putObjectRequest);
197+
PutObjectRequest request = PutObjectRequest.builder().bucket(bucket).key(path).metadata(metadataMap).build();
198+
RequestBody requestBody = RequestBody.fromFile(((TikaInputStream) is).getFile());
199+
s3Client.putObject(request, requestBody);
192200
} catch (IOException e) {
193201
throw new TikaEmitterException("exception sending underlying file", e);
194202
}
195203
return;
196204
}
197205
}
198206
try {
199-
s3Client.putObject(bucket, path, is, objectMetadata);
200-
} catch (AmazonClientException e) {
207+
PutObjectRequest request = PutObjectRequest.builder().bucket(bucket).key(path).metadata(metadataMap).build();
208+
RequestBody requestBody = RequestBody.fromBytes(is.readAllBytes());
209+
s3Client.putObject(request, requestBody);
210+
} catch (S3Exception e) {
201211
throw new IOException("problem writing s3object", e);
202212
}
203213
}
@@ -294,32 +304,38 @@ public void setEndpointConfigurationService(String endpointConfigurationService)
294304
@Override
295305
public void initialize(Map<String, Param> params) throws TikaConfigException {
296306
//params have already been set...ignore them
297-
AWSCredentialsProvider provider;
298-
if ("instance".equals(credentialsProvider)) {
299-
provider = InstanceProfileCredentialsProvider.getInstance();
300-
} else if ("profile".equals(credentialsProvider)) {
301-
provider = new ProfileCredentialsProvider(profile);
302-
} else if (credentialsProvider.equals("key_secret")) {
303-
provider = new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey));
304-
} else {
305-
throw new TikaConfigException("credentialsProvider must be set and " + "must be either 'instance', 'profile' or 'key_secret'");
307+
software.amazon.awssdk.auth.credentials.AwsCredentialsProvider provider;
308+
switch (credentialsProvider)
309+
{
310+
case "instance":
311+
provider = software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider.builder().build();
312+
break;
313+
case "profile":
314+
provider = software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider.builder().profileName(profile).build();
315+
break;
316+
case "key_secret":
317+
AwsBasicCredentials awsCreds = AwsBasicCredentials.create(accessKey, secretKey);
318+
provider = software.amazon.awssdk.auth.credentials.StaticCredentialsProvider.create(awsCreds);
319+
break;
320+
default:
321+
throw new TikaConfigException("credentialsProvider must be set and " + "must be either 'instance', 'profile' or 'key_secret'");
306322
}
307-
ClientConfiguration clientConfig = new ClientConfiguration().withMaxConnections(maxConnections);
308-
try {
309-
AmazonS3ClientBuilder amazonS3ClientBuilder = AmazonS3ClientBuilder
310-
.standard()
311-
.withClientConfiguration(clientConfig)
312-
.withCredentials(provider)
313-
.withPathStyleAccessEnabled(pathStyleAccessEnabled);
314-
if (!StringUtils.isBlank(endpointConfigurationService)) {
315-
amazonS3ClientBuilder.setEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpointConfigurationService, region));
316-
} else {
317-
amazonS3ClientBuilder.withRegion(region);
323+
SdkHttpClient httpClient = ApacheHttpClient.builder().maxConnections(maxConnections).build();
324+
S3Configuration clientConfig1 = S3Configuration.builder().pathStyleAccessEnabled(pathStyleAccessEnabled).build();
325+
S3ClientBuilder s3ClientBuilder = S3Client.builder().httpClient(httpClient).
326+
requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED). // https://stackoverflow.com/a/79488850/535646
327+
serviceConfiguration(clientConfig1).credentialsProvider(provider);
328+
if (!StringUtils.isBlank(endpointConfigurationService)) {
329+
try {
330+
s3ClientBuilder.endpointOverride(new URI(endpointConfigurationService)).region(Region.of(region));
318331
}
319-
s3Client = amazonS3ClientBuilder.build();
320-
} catch (AmazonClientException e) {
321-
throw new TikaConfigException("can't initialize s3 emitter", e);
332+
catch (URISyntaxException ex) {
333+
throw new TikaConfigException("bad endpointConfigurationService: " + endpointConfigurationService, ex);
334+
}
335+
} else {
336+
s3ClientBuilder.region(Region.of(region));
322337
}
338+
s3Client = s3ClientBuilder.build();
323339
}
324340

325341
@Override

0 commit comments

Comments
 (0)