rzachariah
commented
8 years ago Here's the code that creates and sets the spout.
final KinesisSpoutConfig config =
new KinesisSpoutConfig(streamName, zookeeperEndpoint).withZookeeperPrefix(zookeeperPrefix)
.withInitialPositionInStream(initialPositionInStream)
.withRegion(Regions.fromName(regionName));
final KinesisSpout spout = new KinesisSpout(config, new CustomCredentialsProviderChain(), new ClientConfiguration());
TopologyBuilder builder = new TopologyBuilder();
LOG.info("Using Kinesis stream: " + config.getStreamName());
// Using number of shards as the parallelism hint for the spout.
builder.setSpout("Kinesis", spout, 2);Here's the CustomCredentialsProviderChain
/*
* Copyright 2013-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
*
* Licensed under the Amazon Software License (the "License").
* You may not use this file except in compliance with the License.
* A copy of the License is located at
*
* http://aws.amazon.com/asl/
*
* or in the "license" file accompanying this file. This file 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 KinesisStormClickstreamApp;
import com.amazonaws.auth.AWSCredentialsProviderChain;
import com.amazonaws.auth.ClasspathPropertiesFileCredentialsProvider;
import com.amazonaws.auth.EnvironmentVariableCredentialsProvider;
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
import com.amazonaws.auth.SystemPropertiesCredentialsProvider;
/**
* Credentials provider chain that is similar to DefaultAWSCredentialsProviderChain,
* but also includes the ClasspathPropertiesFileCredentialsProvider.
*/
public class CustomCredentialsProviderChain extends AWSCredentialsProviderChain {
public CustomCredentialsProviderChain() {
super(new EnvironmentVariableCredentialsProvider(),
new SystemPropertiesCredentialsProvider(),
new ClasspathPropertiesFileCredentialsProvider("AwsCredentials.properties"),
new InstanceProfileCredentialsProvider());
}
}
I'm encountering an exception when I submit the SampleTopology with a KinesisSpout to a cluster.
The sample is here. https://github.com/awslabs/aws-big-data-blog/tree/master/aws-blog-kinesis-storm-clickstream-app