luciano/amazon-kinesis-client
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

remove sdk v1 dependencies

Browse source
This commit is contained in:
Vincent Viloria 2025-02-27 14:11:03 -08:00
parent f35c7dfd85
commit e72e41c290
13 changed files with 457 additions and 441 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/MultiLangDaemon.java
View file

@ -63,8 +63,8 @@ import software.amazon.kinesis.coordinator.Scheduler;
* # Users can change the credentials provider the KCL will use to retrieve credentials.
* # The DefaultAWSCredentialsProviderChain checks several other providers, which is
* # described here:
* # http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html
* AWSCredentialsProvider = DefaultAWSCredentialsProviderChain
* # https://sdk.amazonaws.com/java/api/2.0.0-preview-11/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html
* AwsCredentialsProvider = DefaultCredentialsProvider
* </pre>
*/
@Slf4j

86
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/auth/KclSTSAssumeRoleSessionCredentialsProvider.java
View file

@ -1,86 +0,0 @@
/*
* Copyright 2023 Amazon.com, Inc. or its affiliates.
* Licensed 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 software.amazon.kinesis.multilang.auth;
import java.util.Arrays;
import com.amazonaws.auth.AWSSessionCredentials;
import com.amazonaws.auth.AWSSessionCredentialsProvider;
import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider.Builder;
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration;
import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClient;
import software.amazon.kinesis.multilang.NestedPropertyKey;
import software.amazon.kinesis.multilang.NestedPropertyProcessor;
import software.amazon.awssdk.regions.Region;
/**
* An {@link AWSSessionCredentialsProvider} that is backed by STSAssumeRole.
*/
public class KclSTSAssumeRoleSessionCredentialsProvider
implements AWSSessionCredentialsProvider, NestedPropertyProcessor {
private final Builder builder;
private final STSAssumeRoleSessionCredentialsProvider provider;
/**
*
* @param params vararg parameters which must include roleArn at index=0,
* and roleSessionName at index=1
*/
public KclSTSAssumeRoleSessionCredentialsProvider(final String[] params) {
this(params[0], params[1], Arrays.copyOfRange(params, 2, params.length));
}
public KclSTSAssumeRoleSessionCredentialsProvider(
final String roleArn, final String roleSessionName, final String... params) {
builder = new Builder(roleArn, roleSessionName);
NestedPropertyKey.parse(this, params);
provider = builder.build();
}
@Override
public AWSSessionCredentials getCredentials() {
return provider.getCredentials();
}
@Override
public void refresh() {
// do nothing
}
@Override
public void acceptEndpoint(final String serviceEndpoint, final String signingRegion) {
final EndpointConfiguration endpoint = new EndpointConfiguration(serviceEndpoint, signingRegion);
final AWSSecurityTokenService stsClient = AWSSecurityTokenServiceClient.builder()
.withEndpointConfiguration(endpoint)
.build();
builder.withStsClient(stsClient);
}
@Override
public void acceptEndpointRegion(final Region region) {
final AWSSecurityTokenService stsClient =
AWSSecurityTokenServiceClient.builder().withRegion(region).build();
builder.withStsClient(stsClient);
}
@Override
public void acceptExternalId(final String externalId) {
builder.withExternalId(externalId);
}
}

59
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/auth/KclStsAssumeRoleCredentialsProvider.java Normal file
View file

@ -0,0 +1,59 @@
package software.amazon.kinesis.multilang.auth;
import java.net.URI;
import java.util.Arrays;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sts.StsClient;
import software.amazon.awssdk.services.sts.StsClientBuilder;
import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
import software.amazon.awssdk.services.sts.model.AssumeRoleRequest.Builder;
import software.amazon.kinesis.multilang.NestedPropertyKey;
import software.amazon.kinesis.multilang.NestedPropertyProcessor;
public class KclStsAssumeRoleCredentialsProvider implements AwsCredentialsProvider, NestedPropertyProcessor {
private final Builder assumeRoleRequestBuilder;
private final StsClientBuilder stsClientBuilder;
public KclStsAssumeRoleCredentialsProvider(String[] params) {
this(params[0], params[1], Arrays.copyOfRange(params, 2, params.length));
}
public KclStsAssumeRoleCredentialsProvider(String roleArn, String roleSessionName, String... params) {
this.assumeRoleRequestBuilder =
AssumeRoleRequest.builder().roleArn(roleArn).roleSessionName(roleSessionName);
this.stsClientBuilder = StsClient.builder();
NestedPropertyKey.parse(this, params);
}
@Override
public AwsCredentials resolveCredentials() {
StsAssumeRoleCredentialsProvider provider = StsAssumeRoleCredentialsProvider.builder()
.refreshRequest(assumeRoleRequestBuilder.build())
.stsClient(stsClientBuilder.build())
.build();
return provider.resolveCredentials();
}
@Override
public void acceptEndpoint(String serviceEndpoint, String signingRegion) {
if (!serviceEndpoint.startsWith("http://") && !serviceEndpoint.startsWith("https://")) {
serviceEndpoint = "https://" + serviceEndpoint;
}
stsClientBuilder.endpointOverride(URI.create(serviceEndpoint));
stsClientBuilder.region(Region.of(signingRegion));
}
@Override
public void acceptEndpointRegion(Region region) {
stsClientBuilder.region(region);
}
@Override
public void acceptExternalId(String externalId) {
assumeRoleRequestBuilder.externalId(externalId);
}
}

182
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/AWSCredentialsProviderPropertyValueDecoder.java
View file

@ -1,182 +0,0 @@
/*
* Copyright 2019 Amazon.com, Inc. or its affiliates.
* Licensed 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 software.amazon.kinesis.multilang.config;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.AWSCredentialsProviderChain;
import lombok.extern.slf4j.Slf4j;
/**
* Get AWSCredentialsProvider property.
*/
@Slf4j
class AWSCredentialsProviderPropertyValueDecoder implements IPropertyValueDecoder<AWSCredentialsProvider> {
private static final String LIST_DELIMITER = ",";
private static final String ARG_DELIMITER = "|";
/**
* Constructor.
*/
AWSCredentialsProviderPropertyValueDecoder() {}
/**
* Get AWSCredentialsProvider property.
*
* @param value
* property value as String
* @return corresponding variable in correct type
*/
@Override
public AWSCredentialsProvider decodeValue(String value) {
if (value != null) {
List<String> providerNames = getProviderNames(value);
List<AWSCredentialsProvider> providers = getValidCredentialsProviders(providerNames);
AWSCredentialsProvider[] ps = new AWSCredentialsProvider[providers.size()];
providers.toArray(ps);
return new AWSCredentialsProviderChain(providers);
} else {
throw new IllegalArgumentException("Property AWSCredentialsProvider is missing.");
}
}
/**
* @return list of supported types
*/
@Override
public List<Class<AWSCredentialsProvider>> getSupportedTypes() {
return Collections.singletonList(AWSCredentialsProvider.class);
}
/**
* Convert string list to a list of valid credentials providers.
*/
private static List<AWSCredentialsProvider> getValidCredentialsProviders(List<String> providerNames) {
List<AWSCredentialsProvider> credentialsProviders = new ArrayList<>();
for (String providerName : providerNames) {
final String[] nameAndArgs = providerName.split("\\" + ARG_DELIMITER);
final Class<? extends AWSCredentialsProvider> clazz;
try {
final Class<?> c = Class.forName(nameAndArgs[0]);
if (!AWSCredentialsProvider.class.isAssignableFrom(c)) {
continue;
}
clazz = (Class<? extends AWSCredentialsProvider>) c;
} catch (ClassNotFoundException cnfe) {
// Providers are a product of prefixed Strings to cover multiple
// namespaces (e.g., "Foo" -> { "some.auth.Foo", "kcl.auth.Foo" }).
// It's expected that many class names will not resolve.
continue;
}
log.info("Attempting to construct {}", clazz);
AWSCredentialsProvider provider = null;
if (nameAndArgs.length > 1) {
final String[] varargs = Arrays.copyOfRange(nameAndArgs, 1, nameAndArgs.length);
// attempt to invoke an explicit N-arg constructor of FooClass(String, String, ...)
provider = constructProvider(providerName, () -> {
Class<?>[] argTypes = new Class<?>[nameAndArgs.length - 1];
Arrays.fill(argTypes, String.class);
return clazz.getConstructor(argTypes).newInstance(varargs);
});
if (provider == null) {
// attempt to invoke a public varargs/array constructor of FooClass(String[])
provider = constructProvider(providerName, () -> clazz.getConstructor(String[].class)
.newInstance((Object) varargs));
}
}
if (provider == null) {
// regardless of parameters, fallback to invoke a public no-arg constructor
provider = constructProvider(providerName, clazz::newInstance);
}
if (provider != null) {
credentialsProviders.add(provider);
}
}
return credentialsProviders;
}
private static List<String> getProviderNames(String property) {
// assume list delimiter is ","
String[] elements = property.split(LIST_DELIMITER);
List<String> result = new ArrayList<>();
for (int i = 0; i < elements.length; i++) {
String string = elements[i].trim();
if (!string.isEmpty()) {
// find all possible names and add them to name list
result.addAll(getPossibleFullClassNames(string));
}
}
return result;
}
private static List<String> getPossibleFullClassNames(final String provider) {
return Stream.of(
// Customer provides a short name of common providers in com.amazonaws.auth package
// (e.g., any classes implementing the AWSCredentialsProvider interface)
// @see
// http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html
"com.amazonaws.auth.",
// Customer provides a short name of a provider offered by this multi-lang package
"software.amazon.kinesis.multilang.auth.",
// Customer provides a fully-qualified provider name, or a custom credentials provider
// (e.g., com.amazonaws.auth.ClasspathFileCredentialsProvider, org.mycompany.FooProvider)
"")
.map(prefix -> prefix + provider)
.collect(Collectors.toList());
}
@FunctionalInterface
private interface CredentialsProviderConstructor<T extends AWSCredentialsProvider> {
T construct()
throws IllegalAccessException, InstantiationException, InvocationTargetException, NoSuchMethodException;
}
/**
* Attempts to construct an {@link AWSCredentialsProvider}.
*
* @param providerName Raw, unmodified provider name. Should there be an
* Exeception during construction, this parameter will be logged.
* @param constructor supplier-like function that will perform the construction
* @return the constructed provider, if successful; otherwise, null
*
* @param <T> type of the CredentialsProvider to construct
*/
private static <T extends AWSCredentialsProvider> T constructProvider(
final String providerName, final CredentialsProviderConstructor<T> constructor) {
try {
return constructor.construct();
} catch (NoSuchMethodException ignored) {
// ignore
} catch (IllegalAccessException | InstantiationException | InvocationTargetException | RuntimeException e) {
log.warn("Failed to construct {}", providerName, e);
}
return null;
}
}

261
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/AwsCredentialsProviderPropertyValueDecoder.java Normal file
View file

@ -0,0 +1,261 @@
/*
* Copyright 2019 Amazon.com, Inc. or its affiliates.
* Licensed 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 software.amazon.kinesis.multilang.config;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import lombok.extern.slf4j.Slf4j;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProviderChain;
import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
import software.amazon.kinesis.multilang.auth.KclStsAssumeRoleCredentialsProvider;
/**
* Get AwsCredentialsProvider property.
*/
@Slf4j
class AwsCredentialsProviderPropertyValueDecoder implements IPropertyValueDecoder<AwsCredentialsProvider> {
private static final String LIST_DELIMITER = ",";
private static final String ARG_DELIMITER = "|";
/**
* Constructor.
*/
AwsCredentialsProviderPropertyValueDecoder() {}
/**
* Get AwsCredentialsProvider property.
*
* @param value
* property value as String
* @return corresponding variable in correct type
*/
@Override
public AwsCredentialsProvider decodeValue(String value) {
if (value != null) {
List<String> providerNames = getProviderNames(value);
List<AwsCredentialsProvider> providers = getValidCredentialsProviders(providerNames);
AwsCredentialsProvider[] ps = new AwsCredentialsProvider[providers.size()];
providers.toArray(ps);
if (providers.isEmpty()) {
log.warn("Unable to construct any provider with name {}", value);
log.warn("Please verify that all AwsCredentialsProvider properties are passed correctly");
}
return AwsCredentialsProviderChain.builder()
.credentialsProviders(providers)
.build();
} else {
throw new IllegalArgumentException("Property AwsCredentialsProvider is missing.");
}
}
/**
* @return list of supported types
*/
@Override
public List<Class<AwsCredentialsProvider>> getSupportedTypes() {
return Collections.singletonList(AwsCredentialsProvider.class);
}
/**
* Convert string list to a list of valid credentials providers.
*/
private static List<AwsCredentialsProvider> getValidCredentialsProviders(List<String> providerNames) {
List<AwsCredentialsProvider> credentialsProviders = new ArrayList<>();
for (String providerName : providerNames) {
final String[] nameAndArgs = providerName.split("\\" + ARG_DELIMITER);
final Class<? extends AwsCredentialsProvider> clazz = getClass(nameAndArgs[0]);
if (clazz == null) {
continue;
}
log.info("Attempting to construct {}", clazz);
final String[] varargs =
nameAndArgs.length > 1 ? Arrays.copyOfRange(nameAndArgs, 1, nameAndArgs.length) : new String[0];
AwsCredentialsProvider provider = tryConstructor(providerName, clazz, varargs);
if (provider == null) {
provider = tryCreate(providerName, clazz, varargs);
}
if (provider != null) {
log.info("Provider constructed successfully: {}", provider);
credentialsProviders.add(provider);
}
}
return credentialsProviders;
}
private static AwsCredentialsProvider tryConstructor(
String providerName, Class<? extends AwsCredentialsProvider> clazz, String[] varargs) {
AwsCredentialsProvider provider =
constructProvider(providerName, () -> getConstructorWithVarArgs(clazz, varargs));
if (provider == null) {
provider = constructProvider(providerName, () -> getConstructorWithArgs(clazz, varargs));
}
if (provider == null) {
provider = constructProvider(providerName, clazz::newInstance);
}
return provider;
}
private static AwsCredentialsProvider tryCreate(
String providerName, Class<? extends AwsCredentialsProvider> clazz, String[] varargs) {
AwsCredentialsProvider provider =
constructProvider(providerName, () -> getCreateMethod(clazz, (Object) varargs));
if (provider == null) {
provider = constructProvider(providerName, () -> getCreateMethod(clazz, varargs));
}
if (provider == null) {
provider = constructProvider(providerName, () -> getCreateMethod(clazz));
}
return provider;
}
private static AwsCredentialsProvider getConstructorWithVarArgs(
Class<? extends AwsCredentialsProvider> clazz, String[] varargs) {
try {
return clazz.getConstructor(String[].class).newInstance((Object) varargs);
} catch (Exception e) {
return null;
}
}
private static AwsCredentialsProvider getConstructorWithArgs(
Class<? extends AwsCredentialsProvider> clazz, String[] varargs) {
try {
Class<?>[] argTypes = new Class<?>[varargs.length];
Arrays.fill(argTypes, String.class);
return clazz.getConstructor(argTypes).newInstance((Object[]) varargs);
} catch (Exception e) {
return null;
}
}
private static AwsCredentialsProvider getCreateMethod(
Class<? extends AwsCredentialsProvider> clazz, Object... args) {
try {
Class<?>[] argTypes = new Class<?>[args.length];
for (int i = 0; i < args.length; i++) {
argTypes[i] = args[i].getClass();
}
Method createMethod = clazz.getDeclaredMethod("create", argTypes);
if (Modifier.isStatic(createMethod.getModifiers())) {
return clazz.cast(createMethod.invoke(null, args));
} else {
log.warn("Found non-static create() method in {}", clazz.getName());
}
} catch (NoSuchMethodException e) {
// No matching create method found for class
} catch (Exception e) {
log.warn("Failed to invoke create() method in {}", clazz.getName(), e);
}
return null;
}
/**
* Resolves the class for the given provider name.
*
* @param providerName A string containing the provider name.
*
* @return The Class object representing the resolved AwsCredentialsProvider implementation,
* or null if the class cannot be resolved or does not extend AwsCredentialsProvider.
*/
private static Class<? extends AwsCredentialsProvider> getClass(String providerName) {
// Convert any form of StsAssumeRoleCredentialsProvider string to KclStsAssumeRoleCredentialsProvider
if (providerName.equals(StsAssumeRoleCredentialsProvider.class.getSimpleName())
|| providerName.equals(StsAssumeRoleCredentialsProvider.class.getName())) {
providerName = KclStsAssumeRoleCredentialsProvider.class.getName();
}
try {
final Class<?> c = Class.forName(providerName);
if (!AwsCredentialsProvider.class.isAssignableFrom(c)) {
return null;
}
return (Class<? extends AwsCredentialsProvider>) c;
} catch (ClassNotFoundException cnfe) {
// Providers are a product of prefixed Strings to cover multiple
// namespaces (e.g., "Foo" -> { "some.auth.Foo", "kcl.auth.Foo" }).
// It's expected that many class names will not resolve.
return null;
}
}
private static List<String> getProviderNames(String property) {
// assume list delimiter is ","
String[] elements = property.split(LIST_DELIMITER);
List<String> result = new ArrayList<>();
for (int i = 0; i < elements.length; i++) {
String string = elements[i].trim();
if (!string.isEmpty()) {
// find all possible names and add them to name list
result.addAll(getPossibleFullClassNames(string));
}
}
return result;
}
private static List<String> getPossibleFullClassNames(final String provider) {
return Stream.of(
// Customer provides a short name of a provider offered by this multi-lang package
"software.amazon.kinesis.multilang.auth.",
// Customer provides a short name of common providers in software.amazon.awssdk.auth.credentials
// package (e.g., any classes implementing the AwsCredentialsProvider interface)
// @see
// https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html
"software.amazon.awssdk.auth.credentials.",
// Customer provides a fully-qualified provider name, or a custom credentials provider
// (e.g., org.mycompany.FooProvider)
"")
.map(prefix -> prefix + provider)
.collect(Collectors.toList());
}
@FunctionalInterface
private interface CredentialsProviderConstructor<T extends AwsCredentialsProvider> {
T construct()
throws IllegalAccessException, InstantiationException, InvocationTargetException, NoSuchMethodException;
}
/**
* Attempts to construct an {@link AwsCredentialsProvider}.
*
* @param providerName Raw, unmodified provider name. Should there be an
* Exception during construction, this parameter will be logged.
* @param constructor supplier-like function that will perform the construction
* @return the constructed provider, if successful; otherwise, null
*
* @param <T> type of the CredentialsProvider to construct
*/
private static <T extends AwsCredentialsProvider> T constructProvider(
final String providerName, final CredentialsProviderConstructor<T> constructor) {
try {
return constructor.construct();
} catch (NoSuchMethodException
| IllegalAccessException
| InstantiationException
| InvocationTargetException
| RuntimeException ignored) {
// ignore
}
return null;
}
}

17
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/KinesisClientLibConfigurator.java
View file

@ -28,7 +28,7 @@ import software.amazon.kinesis.common.StreamIdentifier;
/**
* KinesisClientLibConfigurator constructs a KinesisClientLibConfiguration from java properties file. The following
* three properties must be provided. 1) "applicationName" 2) "streamName" 3) "AWSCredentialsProvider"
* three properties must be provided. 1) "applicationName" 2) "streamName" 3) "AwsCredentialsProvider"
* KinesisClientLibConfigurator will help to automatically assign the value of "workerId" if this property is not
* provided. In the specified properties file, any properties, which matches the variable name in
* KinesisClientLibConfiguration and has a corresponding "with{variableName}" setter method, will be read in, and its
@ -62,7 +62,7 @@ public class KinesisClientLibConfigurator {
properties.entrySet().forEach(e -> {
try {
log.info("Processing (key={}, value={})", e.getKey(), e.getValue());
utilsBean.setProperty(configuration, (String) e.getKey(), e.getValue());
utilsBean.setProperty(configuration, processKey((String) e.getKey()), e.getValue());
} catch (IllegalAccessException | InvocationTargetException ex) {
throw new RuntimeException(ex);
}
@ -110,4 +110,17 @@ public class KinesisClientLibConfigurator {
}
return getConfiguration(properties);
}
/**
* Processes a configuration key to normalize AWS credentials provider naming. Necessary to conform to
* autogenerated setters.
* @param key the config param key
* @return case-configured param key name
*/
String processKey(String key) {
if (key.toLowerCase().startsWith("awscredentialsprovider")) {
key = key.replaceAll("(?i)awscredentialsprovider", "awsCredentialsProvider");
}
return key;
}
}

13
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/MultiLangDaemonConfiguration.java
View file

@ -55,7 +55,6 @@ import software.amazon.kinesis.leases.ShardPrioritization;
import software.amazon.kinesis.lifecycle.LifecycleConfig;
import software.amazon.kinesis.metrics.MetricsConfig;
import software.amazon.kinesis.metrics.MetricsLevel;
import software.amazon.kinesis.multilang.config.credentials.V2CredentialWrapper;
import software.amazon.kinesis.processor.ProcessorConfig;
import software.amazon.kinesis.processor.ShardRecordProcessorFactory;
import software.amazon.kinesis.retrieval.RetrievalConfig;
@ -196,19 +195,19 @@ public class MultiLangDaemonConfiguration {
private final BuilderDynaBean kinesisCredentialsProvider;
public void setAWSCredentialsProvider(String providerString) {
public void setAwsCredentialsProvider(String providerString) {
kinesisCredentialsProvider.set("", providerString);
}
private final BuilderDynaBean dynamoDBCredentialsProvider;
public void setAWSCredentialsProviderDynamoDB(String providerString) {
public void setAwsCredentialsProviderDynamoDB(String providerString) {
dynamoDBCredentialsProvider.set("", providerString);
}
private final BuilderDynaBean cloudWatchCredentialsProvider;
public void setAWSCredentialsProviderCloudWatch(String providerString) {
public void setAwsCredentialsProviderCloudWatch(String providerString) {
cloudWatchCredentialsProvider.set("", providerString);
}
@ -282,9 +281,9 @@ public class MultiLangDaemonConfiguration {
ArrayConverter arrayConverter = new ArrayConverter(String[].class, new StringConverter());
arrayConverter.setDelimiter(',');
convertUtilsBean.register(arrayConverter, String[].class);
AWSCredentialsProviderPropertyValueDecoder oldCredentialsDecoder =
new AWSCredentialsProviderPropertyValueDecoder();
Function<String, ?> converter = s -> new V2CredentialWrapper(oldCredentialsDecoder.decodeValue(s));
AwsCredentialsProviderPropertyValueDecoder credentialsDecoder =
new AwsCredentialsProviderPropertyValueDecoder();
Function<String, ?> converter = credentialsDecoder::decodeValue;
this.kinesisCredentialsProvider = new BuilderDynaBean(
AwsCredentialsProvider.class, convertUtilsBean, converter, CREDENTIALS_DEFAULT_SEARCH_PATH);

52
amazon-kinesis-client-multilang/src/main/java/software/amazon/kinesis/multilang/config/credentials/V2CredentialWrapper.java
View file

@ -1,52 +0,0 @@
/*
* Copyright 2019 Amazon.com, Inc. or its affiliates.
* Licensed 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 software.amazon.kinesis.multilang.config.credentials;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.AWSSessionCredentials;
import lombok.RequiredArgsConstructor;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
@RequiredArgsConstructor
public class V2CredentialWrapper implements AwsCredentialsProvider {
private final AWSCredentialsProvider oldCredentialsProvider;
@Override
public AwsCredentials resolveCredentials() {
AWSCredentials current = oldCredentialsProvider.getCredentials();
if (current instanceof AWSSessionCredentials) {
return AwsSessionCredentials.create(
current.getAWSAccessKeyId(),
current.getAWSSecretKey(),
((AWSSessionCredentials) current).getSessionToken());
}
return new AwsCredentials() {
@Override
public String accessKeyId() {
return current.getAWSAccessKeyId();
}
@Override
public String secretAccessKey() {
return current.getAWSSecretKey();
}
};
}
}

4
amazon-kinesis-client-multilang/src/test/java/software/amazon/kinesis/multilang/MultiLangDaemonConfigTest.java
View file

@ -65,7 +65,7 @@ public class MultiLangDaemonConfigTest {
String properties = String.format(
"executableName = %s\n"
+ "applicationName = %s\n"
+ "AWSCredentialsProvider = DefaultAWSCredentialsProviderChain\n"
+ "AwsCredentialsProvider = DefaultCredentialsProvider\n"
+ "processingLanguage = malbolge\n"
+ "regionName = %s\n",
EXE, APPLICATION_NAME, "us-east-1");
@ -182,7 +182,7 @@ public class MultiLangDaemonConfigTest {
@Test
public void testPropertyValidation() {
String propertiesNoExecutableName = "applicationName = testApp \n" + "streamName = fakeStream \n"
+ "AWSCredentialsProvider = DefaultAWSCredentialsProviderChain\n" + "processingLanguage = malbolge";
+ "AwsCredentialsProvider = DefaultCredentialsProvider\n" + "processingLanguage = malbolge";
ClassLoader classLoader = Mockito.mock(ClassLoader.class);
Mockito.doReturn(new ByteArrayInputStream(propertiesNoExecutableName.getBytes()))

7
amazon-kinesis-client-multilang/src/test/java/software/amazon/kinesis/multilang/auth/KclSTSAssumeRoleSessionCredentialsProviderTest.java
View file

@ -31,15 +31,14 @@ public class KclSTSAssumeRoleSessionCredentialsProviderTest {
*/
@Test
public void testConstructorWithoutOptionalParams() {
new KclSTSAssumeRoleSessionCredentialsProvider(new String[] {ARN, SESSION_NAME});
new KclStsAssumeRoleCredentialsProvider(new String[] {ARN, SESSION_NAME});
}
@Test
public void testAcceptEndpoint() {
// discovered exception during e2e testing; therefore, this test is
// to simply verify the constructed STS client doesn't go *boom*
final KclSTSAssumeRoleSessionCredentialsProvider provider =
new KclSTSAssumeRoleSessionCredentialsProvider(ARN, SESSION_NAME);
final KclStsAssumeRoleCredentialsProvider provider = new KclStsAssumeRoleCredentialsProvider(ARN, SESSION_NAME);
provider.acceptEndpoint("endpoint", "us-east-1");
}
@ -53,7 +52,7 @@ public class KclSTSAssumeRoleSessionCredentialsProviderTest {
}
}
private static class VarArgsSpy extends KclSTSAssumeRoleSessionCredentialsProvider {
private static class VarArgsSpy extends KclStsAssumeRoleCredentialsProvider {
private String externalId;

73
amazon-kinesis-client-multilang/src/test/java/software/amazon/kinesis/multilang/config/AWSCredentialsProviderPropertyValueDecoderTest.java
View file

@ -16,16 +16,16 @@ package software.amazon.kinesis.multilang.config;
import java.util.Arrays;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.AWSCredentialsProviderChain;
import com.amazonaws.auth.BasicAWSCredentials;
import lombok.ToString;
import org.hamcrest.Description;
import org.hamcrest.Matcher;
import org.hamcrest.TypeSafeDiagnosingMatcher;
import org.junit.Test;
import software.amazon.kinesis.multilang.auth.KclSTSAssumeRoleSessionCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProviderChain;
import software.amazon.kinesis.multilang.auth.KclStsAssumeRoleCredentialsProvider;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.instanceOf;
@ -40,10 +40,10 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
private final String credentialName1 = AlwaysSucceedCredentialsProvider.class.getName();
private final String credentialName2 = ConstructorCredentialsProvider.class.getName();
private final AWSCredentialsProviderPropertyValueDecoder decoder = new AWSCredentialsProviderPropertyValueDecoder();
private final AwsCredentialsProviderPropertyValueDecoder decoder = new AwsCredentialsProviderPropertyValueDecoder();
@ToString
private static class AWSCredentialsMatcher extends TypeSafeDiagnosingMatcher<AWSCredentialsProvider> {
private static class AWSCredentialsMatcher extends TypeSafeDiagnosingMatcher<AwsCredentialsProvider> {
private final Matcher<String> akidMatcher;
private final Matcher<String> secretMatcher;
@ -52,12 +52,12 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
public AWSCredentialsMatcher(String akid, String secret) {
this.akidMatcher = equalTo(akid);
this.secretMatcher = equalTo(secret);
this.classMatcher = instanceOf(AWSCredentialsProviderChain.class);
this.classMatcher = instanceOf(AwsCredentialsProviderChain.class);
}
@Override
protected boolean matchesSafely(AWSCredentialsProvider item, Description mismatchDescription) {
AWSCredentials actual = item.getCredentials();
protected boolean matchesSafely(AwsCredentialsProvider item, Description mismatchDescription) {
AwsCredentials actual = item.resolveCredentials();
boolean matched = true;
if (!classMatcher.matches(item)) {
@ -65,12 +65,12 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
matched = false;
}
if (!akidMatcher.matches(actual.getAWSAccessKeyId())) {
akidMatcher.describeMismatch(actual.getAWSAccessKeyId(), mismatchDescription);
if (!akidMatcher.matches(actual.accessKeyId())) {
akidMatcher.describeMismatch(actual.accessKeyId(), mismatchDescription);
matched = false;
}
if (!secretMatcher.matches(actual.getAWSSecretKey())) {
secretMatcher.describeMismatch(actual.getAWSSecretKey(), mismatchDescription);
if (!secretMatcher.matches(actual.secretAccessKey())) {
secretMatcher.describeMismatch(actual.secretAccessKey(), mismatchDescription);
matched = false;
}
return matched;
@ -90,25 +90,25 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
@Test
public void testSingleProvider() {
AWSCredentialsProvider provider = decoder.decodeValue(credentialName1);
AwsCredentialsProvider provider = decoder.decodeValue(credentialName1);
assertThat(provider, hasCredentials(TEST_ACCESS_KEY_ID, TEST_SECRET_KEY));
}
@Test
public void testTwoProviders() {
AWSCredentialsProvider provider = decoder.decodeValue(credentialName1 + "," + credentialName1);
AwsCredentialsProvider provider = decoder.decodeValue(credentialName1 + "," + credentialName1);
assertThat(provider, hasCredentials(TEST_ACCESS_KEY_ID, TEST_SECRET_KEY));
}
@Test
public void testProfileProviderWithOneArg() {
AWSCredentialsProvider provider = decoder.decodeValue(credentialName2 + "|arg");
AwsCredentialsProvider provider = decoder.decodeValue(credentialName2 + "|arg");
assertThat(provider, hasCredentials("arg", "blank"));
}
@Test
public void testProfileProviderWithTwoArgs() {
AWSCredentialsProvider provider = decoder.decodeValue(credentialName2 + "|arg1|arg2");
AwsCredentialsProvider provider = decoder.decodeValue(credentialName2 + "|arg1|arg2");
assertThat(provider, hasCredentials("arg1", "arg2"));
}
@ -118,10 +118,10 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
@Test
public void testKclAuthProvider() {
for (final String className : Arrays.asList(
KclSTSAssumeRoleSessionCredentialsProvider.class.getName(), // fully-qualified name
KclSTSAssumeRoleSessionCredentialsProvider.class.getSimpleName() // name-only; needs prefix
KclStsAssumeRoleCredentialsProvider.class.getName(), // fully-qualified name
KclStsAssumeRoleCredentialsProvider.class.getSimpleName() // name-only; needs prefix
)) {
final AWSCredentialsProvider provider = decoder.decodeValue(className + "|arn|sessionName");
final AwsCredentialsProvider provider = decoder.decodeValue(className + "|arn|sessionName");
assertNotNull(className, provider);
}
}
@ -135,28 +135,25 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
final String className = VarArgCredentialsProvider.class.getName();
final String encodedValue = className + "|" + String.join("|", args);
final AWSCredentialsProvider provider = decoder.decodeValue(encodedValue);
assertEquals(Arrays.toString(args), provider.getCredentials().getAWSAccessKeyId());
final AwsCredentialsProvider provider = decoder.decodeValue(encodedValue);
assertEquals(Arrays.toString(args), provider.resolveCredentials().accessKeyId());
}
/**
* This credentials provider will always succeed
*/
public static class AlwaysSucceedCredentialsProvider implements AWSCredentialsProvider {
public static class AlwaysSucceedCredentialsProvider implements AwsCredentialsProvider {
@Override
public AWSCredentials getCredentials() {
return new BasicAWSCredentials(TEST_ACCESS_KEY_ID, TEST_SECRET_KEY);
public AwsCredentials resolveCredentials() {
return AwsBasicCredentials.create(TEST_ACCESS_KEY_ID, TEST_SECRET_KEY);
}
@Override
public void refresh() {}
}
/**
* This credentials provider needs a constructor call to instantiate it
*/
public static class ConstructorCredentialsProvider implements AWSCredentialsProvider {
public static class ConstructorCredentialsProvider implements AwsCredentialsProvider {
private String arg1;
private String arg2;
@ -172,15 +169,12 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
}
@Override
public AWSCredentials getCredentials() {
return new BasicAWSCredentials(arg1, arg2);
public AwsCredentials resolveCredentials() {
return AwsBasicCredentials.create(arg1, arg2);
}
@Override
public void refresh() {}
}
private static class VarArgCredentialsProvider implements AWSCredentialsProvider {
private static class VarArgCredentialsProvider implements AwsCredentialsProvider {
private final String[] args;
@ -189,13 +183,10 @@ public class AWSCredentialsProviderPropertyValueDecoderTest {
}
@Override
public AWSCredentials getCredentials() {
public AwsCredentials resolveCredentials() {
// KISS solution to surface the constructor args
final String flattenedArgs = Arrays.toString(args);
return new BasicAWSCredentials(flattenedArgs, flattenedArgs);
return AwsBasicCredentials.create(flattenedArgs, flattenedArgs);
}
@Override
public void refresh() {}
}
}

132
amazon-kinesis-client-multilang/src/test/java/software/amazon/kinesis/multilang/config/KinesisClientLibConfiguratorTest.java
View file

@ -22,15 +22,14 @@ import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.google.common.collect.ImmutableSet;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.runners.MockitoJUnitRunner;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.kinesis.common.InitialPositionInStream;
import software.amazon.kinesis.metrics.MetricsLevel;
@ -60,7 +59,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = " + credentialName1,
"AwsCredentialsProvider = " + credentialName1,
"workerId = 123"
},
'\n'));
@ -77,7 +76,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"applicationName = app",
"streamName = 123",
"AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"AwsCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"workerId = 123",
"failoverTimeMillis = 100",
"shardSyncIntervalMillis = 500"
@ -98,7 +97,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"applicationName = app",
"streamName = 123",
"AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"AwsCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"initialPositionInStreamExtended = " + epochTimeInSeconds
},
'\n'));
@ -116,7 +115,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"applicationName = app",
"streamName = 123",
"AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"AwsCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"initialPositionInStream = AT_TIMESTAMP"
},
'\n'));
@ -136,7 +135,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"applicationName = app",
"streamName = 123",
"AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"AwsCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"initialPositionInStreamExtended = null"
},
'\n'));
@ -151,7 +150,7 @@ public class KinesisClientLibConfiguratorTest {
public void testWithUnsupportedClientConfigurationVariables() {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(
new String[] {
"AWSCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"AwsCredentialsProvider = " + credentialName1 + ", " + credentialName2,
"workerId = id",
"kinesisClientConfig = {}",
"streamName = stream",
@ -170,7 +169,7 @@ public class KinesisClientLibConfiguratorTest {
MultiLangDaemonConfiguration config = getConfiguration(StringUtils.join(
new String[] {
"streamName = kinesis",
"AWSCredentialsProvider = " + credentialName2 + ", " + credentialName1,
"AwsCredentialsProvider = " + credentialName2 + ", " + credentialName1,
"workerId = w123",
"maxRecords = 10",
"metricsMaxQueueSize = 20",
@ -195,7 +194,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = ABCD, " + credentialName1,
"AwsCredentialsProvider = ABCD, " + credentialName1,
"workerId = 0",
"cleanupLeasesUponShardCompletion = false",
"validateSequenceNumberBeforeCheckpointing = true"
@ -233,7 +232,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = ABCD," + credentialName1,
"AwsCredentialsProvider = ABCD," + credentialName1,
"workerId = 1",
"metricsEnabledDimensions = ShardId, WorkerIdentifier"
},
@ -253,7 +252,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = ABCD," + credentialName1,
"AwsCredentialsProvider = ABCD," + credentialName1,
"workerId = 123",
"initialPositionInStream = TriM_Horizon"
},
@ -268,7 +267,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = ABCD," + credentialName1,
"AwsCredentialsProvider = ABCD," + credentialName1,
"workerId = 123",
"initialPositionInStream = LateSt"
},
@ -283,7 +282,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = ABCD," + credentialName1,
"AwsCredentialsProvider = ABCD," + credentialName1,
"workerId = 123",
"initialPositionInStream = TriM_Horizon",
"abc = 1"
@ -302,7 +301,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = ABCD," + credentialName1,
"AwsCredentialsProvider = ABCD," + credentialName1,
"workerId = 123",
"initialPositionInStream = TriM_Horizon",
"maxGetRecordsThreadPool = 1"
@ -318,7 +317,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = ABCD," + credentialName1,
"AwsCredentialsProvider = ABCD," + credentialName1,
"workerId = 123",
"initialPositionInStream = TriM_Horizon",
"maxGetRecordsThreadPool = 0",
@ -334,7 +333,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = " + credentialName1,
"AwsCredentialsProvider = " + credentialName1,
"workerId = 123",
"failoverTimeMillis = 100nf"
},
@ -348,7 +347,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = " + credentialName1,
"AwsCredentialsProvider = " + credentialName1,
"workerId = 123",
"failoverTimeMillis = -12"
},
@ -380,7 +379,7 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = " + credentialName1,
"AwsCredentialsProvider = " + credentialName1,
"failoverTimeMillis = 100",
"shardSyncIntervalMillis = 500"
},
@ -397,7 +396,7 @@ public class KinesisClientLibConfiguratorTest {
String test = StringUtils.join(
new String[] {
"applicationName = b",
"AWSCredentialsProvider = " + credentialName1,
"AwsCredentialsProvider = " + credentialName1,
"workerId = 123",
"failoverTimeMillis = 100"
},
@ -410,7 +409,7 @@ public class KinesisClientLibConfiguratorTest {
String test = StringUtils.join(
new String[] {
"applicationName = b",
"AWSCredentialsProvider = " + credentialName1,
"AwsCredentialsProvider = " + credentialName1,
"workerId = 123",
"failoverTimeMillis = 100",
"streamName = ",
@ -425,7 +424,7 @@ public class KinesisClientLibConfiguratorTest {
String test = StringUtils.join(
new String[] {
"streamName = a",
"AWSCredentialsProvider = " + credentialName1,
"AwsCredentialsProvider = " + credentialName1,
"workerId = 123",
"failoverTimeMillis = 100"
},
@ -434,12 +433,12 @@ public class KinesisClientLibConfiguratorTest {
}
@Test
public void testWithAWSCredentialsFailed() {
public void testWithAwsCredentialsFailed() {
String test = StringUtils.join(
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = " + credentialName2,
"AwsCredentialsProvider = " + credentialName2,
"failoverTimeMillis = 100",
"shardSyncIntervalMillis = 500"
},
@ -457,6 +456,34 @@ public class KinesisClientLibConfiguratorTest {
}
}
@Test
public void testProcessKeyWithExpectedCasing() {
String key = "AwsCredentialsProvider";
String result = configurator.processKey(key);
assertEquals("awsCredentialsProvider", result);
}
@Test
public void testProcessKeyWithOldCasing() {
String key = "AWSCredentialsProvider";
String result = configurator.processKey(key);
assertEquals("awsCredentialsProvider", result);
}
@Test
public void testProcessKeyWithMixedCasing() {
String key = "AwScReDeNtIaLsPrOvIdEr";
String result = configurator.processKey(key);
assertEquals("awsCredentialsProvider", result);
}
@Test
public void testProcessKeyWithSuffix() {
String key = "awscredentialsproviderDynamoDB";
String result = configurator.processKey(key);
assertEquals("awsCredentialsProviderDynamoDB", result);
}
// TODO: fix this test
@Test
public void testWithDifferentAWSCredentialsForDynamoDBAndCloudWatch() {
@ -464,9 +491,9 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = " + credentialNameKinesis,
"AWSCredentialsProviderDynamoDB = " + credentialNameDynamoDB,
"AWSCredentialsProviderCloudWatch = " + credentialNameCloudWatch,
"AwsCredentialsProvider = " + credentialNameKinesis,
"AwsCredentialsProviderDynamoDB = " + credentialNameDynamoDB,
"AwsCredentialsProviderCloudWatch = " + credentialNameCloudWatch,
"failoverTimeMillis = 100",
"shardSyncIntervalMillis = 500"
},
@ -492,9 +519,9 @@ public class KinesisClientLibConfiguratorTest {
new String[] {
"streamName = a",
"applicationName = b",
"AWSCredentialsProvider = " + credentialNameKinesis,
"AWSCredentialsProviderDynamoDB = " + credentialName2,
"AWSCredentialsProviderCloudWatch = " + credentialName2,
"AwsCredentialsProvider = " + credentialNameKinesis,
"AwsCredentialsProviderDynamoDB = " + credentialName2,
"AwsCredentialsProviderCloudWatch = " + credentialName2,
"failoverTimeMillis = 100",
"shardSyncIntervalMillis = 500"
},
@ -526,71 +553,56 @@ public class KinesisClientLibConfiguratorTest {
/**
* This credentials provider will always succeed
*/
public static class AlwaysSucceedCredentialsProvider implements AWSCredentialsProvider {
public static class AlwaysSucceedCredentialsProvider implements AwsCredentialsProvider {
@Override
public AWSCredentials getCredentials() {
return new BasicAWSCredentials("a", "b");
public AwsCredentials resolveCredentials() {
return AwsBasicCredentials.create("a", "b");
}
@Override
public void refresh() {}
}
/**
* This credentials provider will always succeed
*/
public static class AlwaysSucceedCredentialsProviderKinesis implements AWSCredentialsProvider {
public static class AlwaysSucceedCredentialsProviderKinesis implements AwsCredentialsProvider {
@Override
public AWSCredentials getCredentials() {
return new BasicAWSCredentials("", "");
public AwsCredentials resolveCredentials() {
return AwsBasicCredentials.create("DUMMY_ACCESS_KEY_ID", "DUMMY_SECRET_ACCESS_KEY");
}
@Override
public void refresh() {}
}
/**
* This credentials provider will always succeed
*/
public static class AlwaysSucceedCredentialsProviderDynamoDB implements AWSCredentialsProvider {
public static class AlwaysSucceedCredentialsProviderDynamoDB implements AwsCredentialsProvider {
@Override
public AWSCredentials getCredentials() {
return new BasicAWSCredentials("", "");
public AwsCredentials resolveCredentials() {
return AwsBasicCredentials.create("DUMMY_ACCESS_KEY_ID", "DUMMY_SECRET_ACCESS_KEY");
}
@Override
public void refresh() {}
}
/**
* This credentials provider will always succeed
*/
public static class AlwaysSucceedCredentialsProviderCloudWatch implements AWSCredentialsProvider {
public static class AlwaysSucceedCredentialsProviderCloudWatch implements AwsCredentialsProvider {
@Override
public AWSCredentials getCredentials() {
return new BasicAWSCredentials("", "");
public AwsCredentials resolveCredentials() {
return AwsBasicCredentials.create("DUMMY_ACCESS_KEY_ID", "DUMMY_SECRET_ACCESS_KEY");
}
@Override
public void refresh() {}
}
/**
* This credentials provider will always fail
*/
public static class AlwaysFailCredentialsProvider implements AWSCredentialsProvider {
public static class AlwaysFailCredentialsProvider implements AwsCredentialsProvider {
@Override
public AWSCredentials getCredentials() {
public AwsCredentials resolveCredentials() {
throw new IllegalArgumentException();
}
@Override
public void refresh() {}
}
private MultiLangDaemonConfiguration getConfiguration(String configString) {

8
amazon-kinesis-client-multilang/src/test/resources/multilang.properties
View file

@ -17,10 +17,12 @@ streamName = kclpysample
applicationName = MultiLangTest
# Users can change the credentials provider the KCL will use to retrieve credentials.
# The DefaultAWSCredentialsProviderChain checks several other providers, which is
# Expected key name (case-sensitive):
# AwsCredentialsProvider / AwsCredentialsProviderDynamoDB / AwsCredentialsProviderCloudWatch
# The DefaultCredentialsProvider checks several other providers, which is
# described here:
# http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/DefaultAWSCredentialsProviderChain.html
AWSCredentialsProvider = DefaultAWSCredentialsProviderChain
# https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html
AwsCredentialsProvider = DefaultCredentialsProvider
# Appended to the user agent of the KCL. Does not impact the functionality of the
# KCL in any other way.