title: Amazon Kinesis GetRecords Api Best approach
body: <p>I have setup Kinesis stream in Amazon WebServices. I Also want to accomplish the following tasks:</p>
  
  <ol>
  <li>Put Records into Single Stream with Single Shard (C# Api) - SUCCESS</li>
  <li>Also I wrote Sample App in which multiple Producers are working on different Stream - SUCCESS</li>
  <li>Also I setup Sample App to Perform Mutiple Workers put the Data into Single Stream - SUCCESS</li>
  </ol>
  
  <p>Also I want to be able to enforce the SequenceNumberOrdering in the Reacords.</p>
  
  <p>But the real pain is the GetRecords Consumer Operation using Kinesis C# Api.</p>
  
  <p>I created a sample App for the Records.  The problem is that it doesn't stop the Iteration even if there are no Records present in the Kinesis Stream. Also keeping the SequenceNumber in the DB or some file and retrieving the file again is time consuming - what is the advantage of using Kinesis Stream for GetRecords? </p>
  
  <p>Why does it keep on iterating even when there is no data in the Stream? </p>
  
  <p>I used following piece of code for the REFERENCE;</p>
  
  <pre><code>  private static void GetFilesKinesisStream()
          {
              IAmazonKinesis kinesis = AWSClientFactory.CreateAmazonKinesisClient();
              try
              {
                  ListStreamsResponse listStreams = kinesis.ListStreams();
                  int numBuckets = 0;
                  if (listStreams.StreamNames != null &amp;&amp;
                      listStreams.StreamNames.Count &gt; 0)
                  {
                      numBuckets = listStreams.StreamNames.Count;
                      Console.WriteLine("You have " + numBuckets + " Amazon Kinesis Streams.");
                      Console.WriteLine(string.Join(",\n", listStreams.StreamNames.ToArray()));
  
                      DescribeStreamRequest describeRequest = new DescribeStreamRequest();
                      describeRequest.StreamName = "******************";
  
                      DescribeStreamResponse describeResponse = kinesis.DescribeStream(describeRequest);
                      List&lt;Shard&gt; shards = describeResponse.StreamDescription.Shards;
                      foreach (Shard s in shards)
                      {
                          Console.WriteLine("shard: " + s.ShardId);
                      }
  
                      string primaryShardId = shards[0].ShardId;
  
                      GetShardIteratorRequest iteratorRequest = new GetShardIteratorRequest();
                      iteratorRequest.StreamName = "*********************";
                      iteratorRequest.ShardId = primaryShardId;
                      iteratorRequest.ShardIteratorType = ShardIteratorType.AT_SEQUENCE_NUMBER;
                      iteratorRequest.StartingSequenceNumber = "49544005271533118105145368110776211536226129690186743810";
  
                      GetShardIteratorResponse iteratorResponse = kinesis.GetShardIterator(iteratorRequest);
                      string iterator = iteratorResponse.ShardIterator;
  
                      Console.WriteLine("Iterator: " + iterator);
                      //Step #3 - get records in this iterator
                      GetShardRecords(kinesis, iterator);
  
                      Console.WriteLine("All records read.");
                      Console.ReadLine();
                  }
                  // sr.WriteLine("You have " + numBuckets + " Amazon S3 bucket(s).");
              }
              catch (AmazonKinesisException ex)
              {
                  if (ex.ErrorCode != null &amp;&amp; ex.ErrorCode.Equals("AuthFailure"))
                  {
                      Console.WriteLine("The account you are using is not signed up for Amazon EC2.");
                      Console.WriteLine("You can sign up for Amazon EC2 at http://aws.amazon.com/ec2");
                  }
                  else
                  {
                      Console.WriteLine("Caught Exception: " + ex.Message);
                      Console.WriteLine("Response Status Code: " + ex.StatusCode);
                      Console.WriteLine("Error Code: " + ex.ErrorCode);
                      Console.WriteLine("Error Type: " + ex.ErrorType);
                      Console.WriteLine("Request ID: " + ex.RequestId);
                  }
              }
          }
  
          private static void GetShardRecords(IAmazonKinesis client, string iteratorId)
          {
              //create reqest
              GetRecordsRequest getRequest = new GetRecordsRequest();
              getRequest.Limit = 100;
              getRequest.ShardIterator = iteratorId;
  
  
              //call "get" operation and get everything in this shard range
              GetRecordsResponse getResponse = client.GetRecords(getRequest);
              //get reference to next iterator for this shard
              string nextIterator = getResponse.NextShardIterator;
              //retrieve records
              List&lt;Record&gt; records = getResponse.Records;
  
              //print out each record's data value
              foreach (Record r in records)
              {
                  //pull out (JSON) data in this record
                  string s = Encoding.UTF8.GetString(r.Data.ToArray());
                  Console.WriteLine("Record: " + s);
                  Console.WriteLine("Partition Key: " + r.PartitionKey);
              }
  
              if (null != nextIterator)
              {
                  //if there's another iterator, call operation again
                  GetShardRecords(client, nextIterator);
              }
          }
  </code></pre>