sunny1978
commented
8 years ago Temp Fix: 1: Overwritten public class KinesisSpoutMine extends KinesisSpout overridden "open" ad added System.setProperty("aws.accessKeyId", "...key..."); System.setProperty("aws.secretKey", "...keysecret..."); super.open(..)
Next Issue: Spout cannot get messages from Stream. Its able to fetch - but cannot convert it
PUT Logic: PutRecordRequest putRecordRequest = new PutRecordRequest(); putRecordRequest.setStreamName(myStreamName); putRecordRequest.setData(ByteBuffer.wrap(String.format("Kinesis-Test:testData-%d", createTime).getBytes())); putRecordRequest.setPartitionKey(String.format("partitionKey-%d", createTime)); PutRecordResult putRecordResult = client.putRecord(putRecordRequest);
Outstanding Issues: 1: 0 byte files in HDFS 2: java.lang.IllegalArgumentException: message does not exist
Storm-KinesisSpout-getNext... (error in storm cluster) Result: All are 0 Byte Files being written to HDFS [root@sandbox logs]# hadoop fs -ls /user/conflux/storm/kinesishdfs Found 7 items -rw-r--r-- 3 storm hdfs 0 2016-02-03 01:22 /user/conflux/storm/kinesishdfs/HDFS-1-1-0-1454462371619.txt -rw-r--r-- 3 storm hdfs 0 2016-02-03 01:27 /user/conflux/storm/kinesishdfs/HDFS-1-1-0-1454462605731.txt -rw-r--r-- 3 storm hdfs 0 2016-02-03 01:33 /user/conflux/storm/kinesishdfs/HDFS-1-1-0-1454462874499.txt
2016-02-03 02:24:07 c.a.s.k.s.KinesisShardGetter [INFO] Seeking to com.amazonaws.services.kinesis.stormspout.ShardPosition@edf529e[pos=LATEST,sequenceNum=
2016-02-02 16:15:18 c.a.s.k.s.u.InfiniteConstantBackoffRetry [DEBUG] Caught exception of type com.amazonaws.AmazonClientException, backing off for 1000 ms.
I tested GET and PUT using Streams and Get requests - both worked flawless. I have all 3 variants Batch, Storm and Spark. Spark - used KinesisStreams - working Batch: Can you Get and Put - working Storm: planning to use KinesisSpout library from kinesis. It is failing with no clue.
Code: final KinesisSpoutConfig config = new KinesisSpoutConfig(streamname, zookeeperurl); config.withInitialPositionInStream(ipis); config.withRegion(Regions.fromName(regionName)); config.withCheckpointIntervalMillis(Integer.parseInt(checkinterval)); config.withZookeeperPrefix("kinesis-zooprefix-" + name);
System.setProperty("aws.accessKeyId", key); System.setProperty("aws.secretKey", keysecret); SystemPropertiesCredentialsProvider scp = new SystemPropertiesCredentialsProvider(); final KinesisSpout spout = new KinesisSpoutConflux(config, scp, new ClientConfiguration());
What am I doing wrong?
Storm Logs: 2016-02-02 16:15:17 c.a.s.k.s.KinesisSpout [INFO] KinesisSpoutConflux[taskIndex=0] open() called with topoConfig task index 0 for processing stream Kinesis-Conflux ... 2016-02-02 16:15:17 c.a.s.k.s.KinesisSpout [DEBUG] KinesisSpoutConflux[taskIndex=0] activating. Starting to process stream Kinesis-Test ... 2016-02-02 16:15:17 c.a.s.k.s.KinesisHelper [INFO] Using us-east-1 region
I dont see "nextTuple" getting called.
My Versions: