Fixing Kinesis Sink

[abhishekd0907]

What was the issue?

Kinesis sink was experiencing slowness in writing records as raised in Issue #47

Issue Analysis

Method flushSync was being called at every step of the iterator. flushSync adds a sleep time of at least 500 milliseconds because of its implementation.

public void flushSync() {
        while (getOutstandingRecordsCount() > 0) {
            flush();
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) { }
        }
    }

Proposed Solution

We can flush records before task completion and not at every step of the iterator. These code changes also reduce sleep time and make changes related to error handling.

[itsvikramagr]

@ggeorgiadis - can you please help us review this PR?

[chadlagore]

typo here

[abhishekd0907]

I missed the main change in this PR,

We need to remove the call to flushSync() in the below code which is causing the slowness.

def sendData(partitionKey: String, data: Array[Byte]): String = {
    var sentSeqNumbers = new String
    val future = producer.addUserRecord(streamName, partitionKey, ByteBuffer.wrap(data))
    val kinesisCallBack = new FutureCallback[UserRecordResult]() {

      override def onFailure(t: Throwable): Unit = {
        logError(s"Writing to  $streamName failed due to ${t.getCause}")
        if (failedWrite == null && t!= null) {
          failedWrite = t
          logError(s"Writing to  $streamName failed due to ${t.getCause}")
        }
      }

      override def onSuccess(result: UserRecordResult): Unit = {
        val shardId = result.getShardId
        sentSeqNumbers = result.getSequenceNumber
      }
    }
    Futures.addCallback(future, kinesisCallBack)
    producer.flushSync()
    sentSeqNumbers
  }

Will raise a new PR for this
cc @itsvikramagr