awsIteratedList causes a very tight loop on GetRecord when stream is empty

[ozataman]

If the stream is empty, GetRecordResponse returns an empty list, but a valid starting point for the next request. With a fast connection, this leads to an endless, very high request throughput loop until more data arrive and slow things down. The immediate effect is receiving ThroughputExceeded exceptions with an empty stream.

Not sure what the solution can be, since internals of awsIteratedList is outside of this package's control. The right thing to do may be to remove the IteratedTransaction instance and roll our own iteratedList function that delays by, say, a second if the previous list is empty.

I've already implemented this in our app and can confirm it eliminates the throughput exceptions.

[larskuhtz]

@ozataman could catch. I think that's serious bug. We consider addressing this by changing the IteratedTransaction instance to the following in version 0.2:

-- | The request parameter 'getRecordsLimit' is interpreted as limit for each
-- single request and not for the overall transaction.
--
-- The 'getRecordsResNextShardIterator' is 'Nothing' only if the shard is
-- closed. This instance will terminate the iteration as soon as an empty list
-- of records is returned.
--
instance IteratedTransaction GetRecords GetRecordsResponse where
    nextIteratedRequest GetRecords{..} (GetRecordsResponse _ []) = Nothing
    nextIteratedRequest GetRecords{..} GetRecordsResponse{..} =
        GetRecords getRecordsLimit <$> getRecordsResNextShardIterator

[ozataman]

I think that's certainly better. However, it may still not help the user completely, as even the empty list contains a valid iterator to be used in the next query and it is not possible to extract that from this instance.

It contains a bit of application context, but here is how I've handled it in the interim:

-------------------------------------------------------------------------------
-- | Produce an infinite stream of records from shard.
streamRecords
    :: (Functor n, MonadIO n, MonadReader AppEnv n, MonadCatch n)
    => ShardId
    -> Maybe SequenceNumber
    -> Maybe Int
    -> Producer (ResourceT n) Record
streamRecords sid sn lim = do
    nm <- either (error.toS) id <$> runEitherT getStream
    let pos = case sn of
          Nothing -> Latest
          Just _ -> AfterSequenceNumber
        gsi = GetShardIterator sid pos sn nm
    iter <- lift $ getShardIteratorResShardIterator <$> runKinesis 10 gsi
    go (GetRecords lim iter)
  where

    go r = do
      a <- lift $ runKinesis 10 r
      let rs = getRecordsResRecords a
      unless (null rs) $ C.sourceList rs

      whenJust (nextIteratedRequest r a) $ \ r' -> do
        when (null rs) $ liftIO (threadDelay 1000000)
        go r'

[larskuhtz]

Yes, I agree that your version makes more sense.

The reason for my simple solution above is to make it usable with with the ListResponse instance.

I wonder if in a production setting one may just bypass the IteratedTransaction, in the end it doesn't provide a lot. If we would keep it as is, I think, the ListResponse instance should be deleted. I am fine with either solution.

It would be really cool if Kinesis would support long-polling...