Java 类com.amazonaws.regions.RegionUtils 实例源码

private void configure() {
    Matcher matcher = REGIONAL_ENDPOINT_PATTERN.matcher(uri.toString());
    if (matcher.find()) {
        String bucketName = matcher.group(1);
        String region = matcher.group(2);
        String key = matcher.group(4);
        Region derivedRegion;
        if (region.equals("external-1")) {
            derivedRegion = Region.getRegion(Regions.US_EAST_1);
        } else {
            derivedRegion = RegionUtils.getRegion(region);
        }

        this.region = Optional.of(derivedRegion);
        this.bucketName = bucketName;
        this.key = key;
    } else {
        this.region = Optional.absent();
        this.bucketName = getBucketName(uri.getHost());
        this.key = getS3BucketKey(uri);
    }
}

@Bean
public Region awsRegion() {
    Region region;
    if(regionString != null && !regionString.isEmpty()) {
        region = RegionUtils.getRegion(regionString);
    } else {
        AwsRegionProvider regionProvider = new DefaultAwsRegionProviderChain();
        region = RegionUtils.getRegion(regionProvider.getRegion());
    }

    if(region == null) {
        throw new BeanInitializationException("Unable to determine AWS region");
    }

    return region;
}

public GenericDynamoDB(AmazonDynamoDB client, AWSCredentialsProvider awsCredentials,
                       ClientConfiguration clientConfiguration,
                       SecretsGroupIdentifier groupIdentifier, Class<Entry> clazz, Converters converters,
                       ReadWriteLock readWriteLock) {
    this.clazz = clazz;
    buildMappings();
    this.converters = converters;
    this.awsCredentials = awsCredentials;
    this.clientConfiguration = clientConfiguration;
    this.client = client;
    this.region = RegionUtils.getRegion(groupIdentifier.region.getName());
    this.readWriteLock = readWriteLock;

    RegionLocalResourceName resourceName = new RegionLocalResourceName(groupIdentifier);
    this.tableName = resourceName.toString();
}

/**
 * Specifically made package access for testing.
 * Used for internal consumption of AWS SDK.
 *
 * Tries to determine the service endpoint for the bucket name.
 * Returns the endpoint configured in the client if the region cannot be determined.
 */
URI resolveServiceEndpoint(String bucketName) {

    if (getSignerRegion() != null || isSignerOverridden()) return endpoint;

    final String regionStr = fetchRegionFromCache(bucketName);
    final com.amazonaws.regions.Region region = RegionUtils.getRegion(regionStr);

    if (region == null) {
        log.warn("Region information for "
                + regionStr
                + " is not available. Please upgrade to latest version of AWS Java SDK");
    }

    return region != null
            ? RuntimeHttpUtils.toUri(region.getServiceEndpoint(S3_SERVICE_NAME), clientConfiguration)
            : endpoint;
}

/**
 * Set the request's endpoint and resource path with the new region provided
 *
 * @param request Request to set endpoint for
 * @param regionString  New region to determine endpoint to hit
 */
public void resolveRequestEndpoint(Request<?> request, String regionString) {
    if (regionString != null) {
        final Region r = RegionUtils.getRegion(regionString);

        if (r == null) {
            throw new SdkClientException("Not able to determine region" +
                    " for " + regionString + ".Please upgrade to a newer " +
                    "version of the SDK");
        }

        endpointBuilder.withRegion(r);
    }
    final URI endpoint = endpointBuilder.getServiceEndpoint();
    if (shouldUseVirtualAddressing(endpoint)) {
        request.setEndpoint(convertToVirtualHostEndpoint(endpoint, bucketName));
        request.setResourcePath(SdkHttpUtils.urlEncode(getHostStyleResourcePath(), true));
    } else {
        request.setEndpoint(endpoint);
        if (bucketName != null) {
            request.setResourcePath(SdkHttpUtils.urlEncode(getPathStyleResourcePath(), true));
        }
    }
}

public void createSnapshotFromTagName(TagNameRequest tagNameRequest, Context context) {

        LambdaLogger logger = context.getLogger();
        logger.log("create ebs snapshot from tag name Start. backup target[" + tagNameRequest + "]");

        String regionName = System.getenv("AWS_DEFAULT_REGION");
        AmazonEC2Async client = RegionUtils.getRegion(regionName).createClient(AmazonEC2AsyncClient.class,
                new DefaultAWSCredentialsProviderChain(), cc);

        try {
            List<Volume> volumes = describeBackupVolumes(client, tagNameRequest);

            for (Volume volume : volumes) {
                createSnapshot(volume.getVolumeId(), tagNameRequest.getGenerationCount(), context);
            }
        } finally {
            client.shutdown();
        }
    }

/**
 * Calculates DynamoDB end-point.
 *
 * Algorithm details:
 * <ol>
 * <li> Use endpoint in job configuration "dynamodb.endpoint" value if available
 * <li> Use endpoint from region in job configuration "dynamodb.region" value if available
 * <li> Use endpoint from region in job configuration "dynamodb.regionid" value if available
 * <li> Use endpoint from EC2 Metadata of instance if available
 * <li> If all previous attempts at retrieving endpoint fail, default to us-east-1 endpoint
 * </ol>
 *
 * @param conf   Job Configuration
 * @param region optional preferred region
 * @return end-point for DynamoDb service
 */
public static String getDynamoDBEndpoint(Configuration conf, String region) {
  String endpoint = getValueFromConf(conf, DynamoDBConstants.ENDPOINT);
  if (Strings.isNullOrEmpty(endpoint)) {
    if (Strings.isNullOrEmpty(region)) {
      region = getValueFromConf(conf, DynamoDBConstants.REGION);
    }
    if (Strings.isNullOrEmpty(region)) {
      region = getValueFromConf(conf, DynamoDBConstants.REGION_ID);
    }
    if (Strings.isNullOrEmpty(region)) {
      try {
        region = EC2MetadataUtils.getEC2InstanceRegion();
      } catch (Exception e) {
        log.warn(String.format("Exception when attempting to get AWS region information. Will "
            + "ignore and default " + "to %s", DynamoDBConstants.DEFAULT_AWS_REGION), e);
      }
    }
    if (Strings.isNullOrEmpty(region)) {
      region = DynamoDBConstants.DEFAULT_AWS_REGION;
    }
    endpoint = RegionUtils.getRegion(region).getServiceEndpoint(ServiceAbbreviations.Dynamodb);
  }
  log.info("Using endpoint for DynamoDB: " + endpoint);
  return endpoint;
}

@Override
public AwsToken load(AwsCredentials awsCredentials) throws Exception {
    AmazonECR amazonECR = new AmazonECRClient(new AWSCredentialsProvider() {
        @Override
        public AWSCredentials getCredentials() {
            return awsCredentials;
        }

        @Override
        public void refresh() {
        }
    });
    amazonECR.setRegion(RegionUtils.getRegion(awsCredentials.getRegion()));
    GetAuthorizationTokenResult authorizationToken = amazonECR.getAuthorizationToken(new GetAuthorizationTokenRequest());
    List<AuthorizationData> authorizationData = authorizationToken.getAuthorizationData();
    Assert.isTrue(!CollectionUtils.isEmpty(authorizationData), "authorizationData is null or empty for token " + authorizationToken);
    AuthorizationData data = authorizationData.get(0);
    byte[] decode = Base64.getDecoder().decode(data.getAuthorizationToken());
    String token = new String(decode);
    String[] split = token.split(":");
    log.info("about to connect to AWS endpoint: {}", data.getProxyEndpoint());
    return AwsToken.builder().username(split[0]).password(split[1])
            .expiresAt(data.getExpiresAt()).proxyEndpoint(data.getProxyEndpoint()).build();
}

/**
     * AWS Access Key ID [None]: ACESSKEY
     AWS Secret Access Key [None]: EXAMPLEKEY
     Default region name [None]: us-west-2
     Default output format [None]: ENTER
     */
    @Test

    public void testConnection() {
//        ClientConfiguration configuration = new ClientConfiguration();
//        configuration.set
        amazonECR.setRegion(RegionUtils.getRegion("us-west-2"));
        GetAuthorizationTokenResult authorizationToken = amazonECR.getAuthorizationToken(new GetAuthorizationTokenRequest());
        log.info("authorizationToken: {}", authorizationToken);
        List<AuthorizationData> authorizationData = authorizationToken.getAuthorizationData();

        log.info("token: {}", authorizationData.get(0).getAuthorizationToken());
        log.info("endpoint: {}", authorizationData.get(0).getProxyEndpoint());
//        amazonECR.setEndpoint(authorizationData.get(0).getProxyEndpoint());

//        BatchGetImageResult batchGetImageResult = amazonECR.batchGetImage(new BatchGetImageRequest());
//        log.info("batchGetImageResult {}", batchGetImageResult);
    }

@Before
public void setup() throws Exception {
    kmsClient = mock(AWSKMSClient.class);
    urlResolver = mock(UrlResolver.class);
    lambdaClient = mock(AWSLambdaClient.class);

    mockWebServer = new MockWebServer();
    mockWebServer.start();
    vaultUrl = "http://localhost:" + mockWebServer.getPort();

    when(urlResolver.resolve()).thenReturn(vaultUrl);


    mockStatic(Regions.class);

    when(Regions.getCurrentRegion()).thenReturn(RegionUtils.getRegion("us-west-2"));
    whenNew(AWSLambdaClient.class).withNoArguments().thenReturn(lambdaClient);
    whenNew(AWSKMSClient.class).withAnyArguments().thenReturn(kmsClient);
}

/**
 * Decrypts the given encrypted String for the given encrypted private key.
 *
 * @param encryptedPrivateKeyBytes
 * @param encryptedString
 * @return the decrypted text
 */
public static String decrypt(byte[] encryptedPrivateKeyBytes, Map<String, String> encryptionContext, String encryptedString) throws Exception {

    //Multi region support
    String regionParam = System.getProperty("awssdk.region");
    Region region = regionParam != null ? RegionUtils.getRegion(regionParam) : Regions.getCurrentRegion();

    if (region == null) {
        throw new Exception("Unrecognized AWS region " + regionParam);
    }

    // split and decode the symmetrically-encrypted text from the asymmetrically-encrypted symmetric key
    String[] sections = encryptedString.split(PAYLOAD_SEPARATOR);
    byte[] encryptedSecretKeyBytes = Base64.getDecoder().decode(sections[0]);
    String encryptedText = sections[1];

    // get the decrypted asymmetric private key;
    Cipher privateKeyDecryptionCipher = getPrivateKeyCipher(region, encryptedPrivateKeyBytes, encryptionContext);

    // unwrap the symmetric key from the private key;
    SecretKey secretKey = (SecretKey) privateKeyDecryptionCipher.unwrap(encryptedSecretKeyBytes, "AES", Cipher.SECRET_KEY);

    return Aes.decrypt(secretKey, encryptedText);
}

public JKinesisReceiver(String applicationName, String streamName,
                        String endpointUrl, String regionName,
                        Duration checkpoint, InitialPositionInStream position) {
    super(StorageLevel.MEMORY_ONLY_SER());

    this.workerId = getHostname() + ":" + String.valueOf(UUID.randomUUID());
    this.checkpointInterval = checkpoint;
    this.initialPosition = position;

    Region region = RegionUtils.getRegion(regionName);

    try {
        this.kclConfig = new KinesisClientLibConfiguration(applicationName, streamName,
                                                  getCredsProvider(),
                                                  workerId)
                        .withCommonClientConfig(CLIENT_CONF)
                        .withRegionName(region.getName())
                        .withKinesisEndpoint(endpointUrl)
                        .withInitialPositionInStream(InitialPositionInStream.LATEST)
                        .withTaskBackoffTimeMillis(500);
    } catch (Exception ex) {
        // do absolutely nothing - and feel good about it!
        // but ...
        // we'd do something meaningful in a PROD context
    }
}

public static void main(String[] args) throws Exception {
    verify(args);
    String stream = args[0];
    Region region = RegionUtils.getRegion(args[1]);

    AWSCredentials credentials = getCreds();
    AmazonKinesis client = new AmazonKinesisClient(credentials, CLIENT_CONF);
    client.setRegion(region);
    checkStream(client.describeStream(stream));

    System.out.println("Let's start putting records!");
    Random rnd = new Random(System.currentTimeMillis());
    for (;;) {
        putEventRecord(client, stream);
        Thread.sleep(rnd.nextInt(500) + 650);
    }
}

@ReactMethod
public void initCredentialsProvider(String identityPoolId, String token, String region)
{
    RegionUtils regionUtils = new RegionUtils();
    Region awsRegion = regionUtils.getRegion(region);

    cognitoCredentialsProvider = new CognitoCachingCredentialsProvider(
        mActivityContext.getApplicationContext(),
        identityPoolId,
        // awsRegion);
        Regions.EU_WEST_1);

    cognitoClient = new CognitoSyncManager(
        mActivityContext.getApplicationContext(),
        // awsRegion,
        Regions.EU_WEST_1,
        cognitoCredentialsProvider);
}

@Override
public void start() {
    if (RegionUtils.getRegion(region) == null) {
        addError(format("Region not set or invalid for appender '%s'", getName()));
        return;
    }
    if (StringUtils.isNullOrEmpty(logGroup)) {
        addError(format("Log group name not set for appender '%s'", getName()));
        return;
    }
    if (StringUtils.isNullOrEmpty(logStream)) {
        addError(format("Log stream name not set for appender '%s'", getName()));
        return;
    }
    setConverter(new StringPayloadConverter(getCharset(), isBinary()));
    super.start();
}

@Override
protected void doStart() {
    logs = new AWSLogsClient(
            getCredentials(),
            getClientConfiguration()
    );
    logs.setRegion(RegionUtils.getRegion(region));
    if (!skipCreate) {
        if (!logGroupExists(logGroup)) {
            createLogGroup(logGroup);
        }
        if (!logStreamExists(logGroup, logStream)) {
            createLogStream(logGroup, logStream);
        }
    }
    queue = new LinkedBlockingQueue<>(internalQueueSize);
    worker = new Worker<>(this);
    worker.setName(format("%s-worker", getName()));
    worker.setDaemon(true);
    worker.start();
}

/** Returns an S3 client given the configuration **/
public static AmazonS3Client getS3Client(Map conf) {
    AWSCredentialsProvider provider = new DefaultAWSCredentialsProviderChain();
    AWSCredentials credentials = provider.getCredentials();
    ClientConfiguration config = new ClientConfiguration();

    AmazonS3Client client = new AmazonS3Client(credentials, config);

    String regionName = ConfUtils.getString(conf, REGION);
    if (StringUtils.isNotBlank(regionName)) {
        client.setRegion(RegionUtils.getRegion(regionName));
    }

    String endpoint = ConfUtils.getString(conf, ENDPOINT);
    if (StringUtils.isNotBlank(endpoint)) {
        client.setEndpoint(endpoint);
    }
    return client;
}

@Bean
public SQSConnection sqsConnection(AWSCredentialsProvider awsCredentialsProvider,
                                   ClientConfiguration awsClientConfig) throws JMSException {

    SQSConnectionFactory connectionFactory = SQSConnectionFactory.builder()
            .withRegion(RegionUtils.getRegion(awsRegion))
            .withAWSCredentialsProvider(awsCredentialsProvider)
            .withClientConfiguration(awsClientConfig)
            .build();

    return connectionFactory.createConnection();
}

protected KmsMasterKeyProvider getProvider() {
    if (!prov.isPresent()) {
        Region region = RegionUtils.getRegion(groupIdentifier.region.getName());
        prov = Optional.of(new KmsMasterKeyProvider(awsCredentials, region, transformAndVerifyOrThrow(clientConfiguration), getKeyArn()));
    }
    return prov.get();
}

@Inject
public AWSCucumberStepdefs(AmazonWebServiceClient client) {
    this.client = client;
    this.client.setRegion(RegionUtils.getRegion("us-east-1"));

    Class<?> httpClientClass = Classes.childClassOf(AmazonWebServiceClient.class, this.client);

    this.packageName = httpClientClass.getPackage().getName();
}

@Override
public String getRegionName() {
    String authority = super.endpoint.getAuthority();
    if(Constants.S3_HOSTNAME.equals(authority)) {
        return "us-east-1";
    }
    Matcher m = Region.S3_REGIONAL_ENDPOINT_PATTERN.matcher(authority);
    try {
        m.matches();
        return RegionUtils.getRegion(m.group(1)).getName();
    } catch (Exception e) {
        throw new IllegalStateException("No valid region has been specified. Unable to return region name", e);
    }
}

public List<RegionEntry> getRegions() {
    if(regionEntries != null)
        return regionEntries;

    regionEntries = new ArrayList<>();
    for(Region region : RegionUtils.getRegions()) {
        String description = regionDescriptions.get(region.getName());
        if(description != null)
            regionEntries.add(new RegionEntry(region, description));
    }
    return regionEntries;
}

@Override
public void handleRequest(InputStream is, OutputStream os, Context context) {

    LambdaLogger logger = context.getLogger();
    String regionName = System.getenv("AWS_DEFAULT_REGION");
    AmazonEC2Async client = RegionUtils.getRegion(regionName).createClient(AmazonEC2AsyncClient.class,
            new DefaultAWSCredentialsProviderChain(), cc);
    try {
        ObjectMapper om = new ObjectMapper();
        DeregisterImageRequest event = om.readValue(is, DeregisterImageRequest.class);
        String imageId = event.getDetail().getRequestParameters().getImageId();

        logger.log("Deregister AMI parge snapshot Start. ImageId[" + imageId + "]");

        List<Snapshot> snapshots = describeSnapshot(client, imageId, context);
        if (snapshots.size() == 0) {
            logger.log("Target of snapshot there is nothing.");
        } else {
            snapshots.stream().forEach(s -> pargeSnapshot(client, s.getSnapshotId(), context));
        }
        logger.log("[SUCCESS][DeregisterImage]has been completed successfully." + imageId);
    } catch (Exception e) {
        logger.log("[ERROR][DeregisterImage]An unexpected error has occurred. message[" + e.getMessage() + "]");
    } finally {
        client.shutdown();
    }
}

AWSLogsStub(String logGroupName, String logStreamName, String logRegion) {

        this.logGroupName = logGroupName;
        this.logStreamName = logStreamName;

        AWSLogs awsLogs = new AWSLogsClient();
        if (logRegion != null) {
            awsLogs.setRegion(RegionUtils.getRegion(logRegion));
        }
        this.awsLogs = awsLogs;
    }

@JsonIgnore
public void configure(AmazonDynamoDBAsyncClient client) {
  // set the region or the url, depending on the format
  if (regionOrUrl.contains(":")) {
    client.setEndpoint(regionOrUrl);
  } else {
    client.setRegion(RegionUtils.getRegion(regionOrUrl));
  }
}

private static String getRegionName(String location) {
  final Region region = RegionUtils.getRegion(location);
  if (region == null && location.equals("US")) {
    return Regions.US_EAST_1.getName();
  }

  return region != null ? region.getName() : location;
}

@Override
public void ensureValid(String name, Object region) {
  String regionStr = ((String) region).toLowerCase().trim();
  if (RegionUtils.getRegion(regionStr) == null) {
    throw new ConfigException(
        name,
        region,
        "Value must be one of: " + Utils.join(RegionUtils.getRegions(), ", ")
    );
  }
}

@Before
public void setUp() {
  PowerMockito.spy(RegionUtils.class);
  PowerMockito.mockStatic(EC2MetadataUtils.class);
  region = mock(Region.class);
  conf.clear();
}

@Test
public void getsEndpointFromRegion() {
  when(RegionUtils.getRegion(TEST_REGION)).thenReturn(region);
  when(region.getServiceEndpoint(ServiceAbbreviations.Dynamodb)).thenReturn(TEST_ENDPOINT);
  conf.set(DynamoDBConstants.REGION, TEST_REGION);
  assertEquals(TEST_ENDPOINT, DynamoDBUtil.getDynamoDBEndpoint(conf, null));
  verify(region).getServiceEndpoint(ServiceAbbreviations.Dynamodb);
}

@Test
public void getsEndpointFromEc2InstanceRegion() {
  when(EC2MetadataUtils.getEC2InstanceRegion()).thenReturn("ec2-instance-region");
  when(RegionUtils.getRegion("ec2-instance-region")).thenReturn(region);
  when(region.getServiceEndpoint(ServiceAbbreviations.Dynamodb)).thenReturn(TEST_ENDPOINT);
  assertEquals(TEST_ENDPOINT, DynamoDBUtil.getDynamoDBEndpoint(conf, null));
  PowerMockito.verifyStatic();
  EC2MetadataUtils.getEC2InstanceRegion();
  verify(region, never()).getServiceEndpoint(TEST_REGION);
}

@Test
public void getsEndpointFromDefaultAwsRegion() {
  PowerMockito.mockStatic(RegionUtils.class);
  when(EC2MetadataUtils.getEC2InstanceRegion()).thenThrow(new AmazonClientException("Unable to " +
      "get region from EC2 instance data"));
  when(RegionUtils.getRegion(DynamoDBConstants.DEFAULT_AWS_REGION)).thenReturn(region);
  when(region.getServiceEndpoint(ServiceAbbreviations.Dynamodb)).thenReturn(TEST_ENDPOINT);
  assertEquals(TEST_ENDPOINT, DynamoDBUtil.getDynamoDBEndpoint(conf, null));
  PowerMockito.verifyStatic();
  RegionUtils.getRegion(DynamoDBConstants.DEFAULT_AWS_REGION);
}

/**
 * Looks up the assigned role for the running Lambda via the GetFunctionConfiguration API.  Requests a token from
 * Cerberus and attempts to decrypt it as that role.
 */
@Override
protected void authenticate() {
    final Region currentRegion = RegionUtils.getRegion(this.region);

    final AWSLambda lambdaClient = new AWSLambdaClient();
    lambdaClient.setRegion(currentRegion);

    final GetFunctionConfigurationResult functionConfiguration = lambdaClient.getFunctionConfiguration(
            new GetFunctionConfigurationRequest()
                    .withFunctionName(functionName)
                    .withQualifier(qualifier));

    final String roleArn = functionConfiguration.getRole();

    if (StringUtils.isBlank(roleArn)) {
        throw new IllegalStateException("Lambda function has no assigned role, aborting Cerberus authentication.");
    }

    final Matcher roleArnMatcher = IAM_ROLE_ARN_PATTERN.matcher(roleArn);

    if (!roleArnMatcher.matches()) {
        throw new IllegalStateException("Lambda function assigned role is not a valid IAM role ARN.");
    }

    try {
        getAndSetToken(roleArn, currentRegion);
        return;
    } catch (AmazonClientException ace) {
        LOGGER.warn("Unexpected error communicating with AWS services.", ace);
    } catch (JsonSyntaxException jse) {
        LOGGER.error("The decrypted auth response was not in the expected format!", jse);
    } catch (VaultClientException sce) {
        LOGGER.warn("Unable to acquire Vault token for IAM role: " + roleArn, sce);
    }

    throw new VaultClientException("Unable to acquire token with Lambda instance role.");
}

/**
 * Returns the KMS endpoint URL for the specified region.
 *
 * @param kmsClient  the KMS client
 * @param regionName the desired region
 * @return the endpoint URL for the specified region
 * @throws IllegalArgumentException if the endpoint cannot be determined
 */
private static String getKMSEndpointForRegion(AWSKMSClient kmsClient, String regionName) {
  requireNonNull(kmsClient, "kmsClient is null");
  requireNonNull(regionName, "regionName is null");

  com.amazonaws.regions.Region region = RegionUtils.getRegion(regionName);

  if (region == null) {
    throw new IllegalArgumentException(String.format("Unable to find the region %s", regionName));
  }

  String serviceName = kmsClient.getServiceName();
  String protocolPrefix = region.hasHttpsEndpoint(serviceName) ? "https://" : "http://";
  return protocolPrefix + region.getServiceEndpoint(serviceName);
}

void setSqsRegion() {
    String regionName = this.configuration.getSqsConfiguration().getRegion();
    Region region = RegionUtils.getRegion(regionName);
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Setting SQS region to " + region.getName());
    }
    sqs.setRegion(region);
}

/**
 * Looks up table name and creates one if it does not exist
 */
public UserAuthentication() {
    ddb = new AmazonDynamoDBClient();
    ddb.setRegion(RegionUtils.getRegion(Configuration.REGION));

    try {
        if (!doesTableExist(USER_TABLE)) {
            createIdentityTable();
        }
    } catch (DataAccessException e) {
        throw new RuntimeException("Failed to create device table.", e);
    }
}

/**
 * Looks up table name and creates one if it does not exist
 */
public DeviceAuthentication() {
    ddb = new AmazonDynamoDBClient();
    ddb.setRegion(RegionUtils.getRegion(Configuration.REGION));

    try {
        if (!doesTableExist(DEVICE_TABLE)) {
            createDeviceTable();
        }
    } catch (DataAccessException e) {
        throw new RuntimeException("Failed to create device table.", e);
    }
}

@Test
public void testCustomRegionAndPathStyleAccess()
        throws Exception
{
    Config config = newConfig();
    config.set("path_style_access", true);
    config.set("_command", BUCKET + "/" + KEY);
    when(taskRequest.getConfig()).thenReturn(config);

    when(s3Secrets.getSecretOptional("region")).thenReturn(Optional.of(REGION));

    when(s3Client.getObjectMetadata(objectMetadataRequestCaptor.capture())).thenThrow(NOT_FOUND_EXCEPTION);

    Operator operator = factory.newOperator(newContext(projectPath, taskRequest));

    try {
        operator.run();
        fail();
    }
    catch (TaskExecutionException ignore) {
    }

    verify(s3Client).setS3ClientOptions(s3ClientOptionsCaptor.capture());
    S3ClientOptions s3ClientOptions = s3ClientOptionsCaptor.getValue();
    assertThat(s3ClientOptions.isPathStyleAccess(), is(true));

    verify(s3Client).setRegion(RegionUtils.getRegion(REGION));
}

/**
 * Creates a new StreamSource.
 * 
 * @param config
 *        Configuration to determine which stream to put records to and get {@link AWSCredentialsProvider}
 * @param inputFile
 *        File containing record data to emit on each line
 * @param loopOverStreamSource
 *        Loop over the stream source to continually put records
 */
public StreamSource(KinesisConnectorConfiguration config, String inputFile, boolean loopOverStreamSource) {
    this.config = config;
    this.inputFile = inputFile;
    this.loopOverInputFile = loopOverStreamSource;
    this.objectMapper = new ObjectMapper();
    kinesisClient = new AmazonKinesisClient(config.AWS_CREDENTIALS_PROVIDER);
    kinesisClient.setRegion(RegionUtils.getRegion(config.REGION_NAME));
    if (config.KINESIS_ENDPOINT != null) {
        kinesisClient.setEndpoint(config.KINESIS_ENDPOINT);
    }
    KinesisUtils.createInputStream(config);
}

/**
 * Creates the Amazon Kinesis stream specified by config.KINESIS_INPUT_STREAM
 * 
 * @param config
 *        The configuration with the specified input stream name and {@link AWSCredentialsProvider}
 * @param shardCount
 *        The shard count to create the stream with
 */
public static void createInputStream(KinesisConnectorConfiguration config) {
    AmazonKinesisClient kinesisClient = new AmazonKinesisClient(config.AWS_CREDENTIALS_PROVIDER);
    kinesisClient.setRegion(RegionUtils.getRegion(config.REGION_NAME));
    if (config.KINESIS_ENDPOINT != null) {
        kinesisClient.setEndpoint(config.KINESIS_ENDPOINT);
    }
    createAndWaitForStreamToBecomeAvailable(kinesisClient,
            config.KINESIS_INPUT_STREAM,
            config.KINESIS_INPUT_STREAM_SHARD_COUNT);
}

/**
 * Creates the Amazon Kinesis stream specified by config.KINESIS_OUTPUT_STREAM.
 * 
 * @param config
 *        The configuration with the specified output stream name and {@link AWSCredentialsProvider}
 * @param shardCount
 *        The shard count to create the stream with
 */
public static void createOutputStream(KinesisConnectorConfiguration config) {
    AmazonKinesisClient kinesisClient = new AmazonKinesisClient(config.AWS_CREDENTIALS_PROVIDER);
    kinesisClient.setRegion(RegionUtils.getRegion(config.REGION_NAME));
    if (config.KINESIS_ENDPOINT != null) {
        kinesisClient.setEndpoint(config.KINESIS_ENDPOINT);
    }
    createAndWaitForStreamToBecomeAvailable(kinesisClient,
            config.KINESIS_OUTPUT_STREAM,
            config.KINESIS_OUTPUT_STREAM_SHARD_COUNT);
}