Make scheduleFlush in KinesisSink non-recursive

[dilyand]

This function should be called at the end of flush() and should not call itself.

At the end of each flush(), whatever the trigger, we call this function to schedule a flush after TimeThreshold in case we haven't reached the bytes and records limits.

If these limits have been reached in-between, then flush won't happen because of currentTime - lastFlushed >= TimeThreshold and another flush will have already been scheduled.

def scheduleFlush(interval: Long = TimeThreshold): Unit = {
    executorService.schedule(
      new Thread {
        override def run(): Unit = {
          val lastFlushed = EventStorage.getLastFlushTime
          val currentTime = System.currentTimeMillis()
          if (currentTime - lastFlushed >= TimeThreshold) {
            EventStorage.flush()
            scheduleFlush(TimeThreshold)
          } else {
            scheduleFlush(TimeThreshold + lastFlushed - currentTime)
          }
        }
      },
      interval,
      MILLISECONDS
    )
    ()
  }

def scheduleFlush(): Unit = {
    executorService.schedule(
      new Thread {
        override def run(): Unit = {
          val lastFlushed = EventStorage.getLastFlushTime
          val currentTime = System.currentTimeMillis()
          if (currentTime - lastFlushed >= TimeThreshold) {
            EventStorage.flush()
          }
        }
      },
      TimeThreshold,
      MILLISECONDS
    )
    ()
  }

#124 (comment)