diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,12 @@ +### v0.3.0.0 + +All changes were in the Producer client. + +- Do not retry messages indefinitely; add support for a configurable retry + policy. + +- Reject messages synchronously which exceed the Kinesis record size limit. + ### v0.2.0.3 All changes were in the Consumer CLI. diff --git a/aws-kinesis-client.cabal b/aws-kinesis-client.cabal --- a/aws-kinesis-client.cabal +++ b/aws-kinesis-client.cabal @@ -1,5 +1,5 @@ name: aws-kinesis-client -version: 0.2.0.3 +version: 0.3.0.0 synopsis: A producer & consumer client library for AWS Kinesis -- description: license: Apache-2.0 diff --git a/src/Aws/Kinesis/Client/Producer.hs b/src/Aws/Kinesis/Client/Producer.hs --- a/src/Aws/Kinesis/Client/Producer.hs +++ b/src/Aws/Kinesis/Client/Producer.hs @@ -28,6 +28,7 @@ {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE PatternSynonyms #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE TupleSections #-} @@ -56,6 +57,7 @@ , pkKinesisKit , pkStreamName , pkBatchPolicy +, pkRetryPolicy , pkMessageQueueBounds , pkMaxConcurrency @@ -63,11 +65,19 @@ , _KinesisError , _MessageNotEnqueued , _InvalidConcurrentConsumerCount +, _MessageTooLarge +, pattern MaxMessageSize + , BatchPolicy , defaultBatchPolicy , bpBatchSize , bpEndpoint + +, RetryPolicy +, defaultRetryPolicy +, rpRetryCount + , RecordEndpoint(..) ) where @@ -110,6 +120,10 @@ deriving (Eq, Show) +-- | The maximum size in bytes of a message. +-- +pattern MaxMessageSize = 51000 + -- | The producer batches records according to a user-specified policy. -- data BatchPolicy @@ -129,7 +143,7 @@ bpEndpoint ∷ Lens' BatchPolicy RecordEndpoint bpEndpoint = lens _bpEndpoint $ \bp ep → bp { _bpEndpoint = ep } --- | The default batching policy sends '200' records per 'PutRecordsEndpoint' +-- | The default batching policy sends @200@ records per 'PutRecordsEndpoint' -- request. -- defaultBatchPolicy ∷ BatchPolicy @@ -138,13 +152,39 @@ , _bpEndpoint = PutRecordsEndpoint } +-- | The producer will attempt to re-send records which failed according to a +-- user-specified policy. This policy applies to failures which occur in the +-- process of sending a message to Kinesis, not those which occur in the course +-- of enqueuing a message. +data RetryPolicy + = RetryPolicy + { _rpRetryCount ∷ {-# UNPACK #-} !Int + } deriving (Eq, Show) +-- | The number of times to retry sending a message after it has first failed. +-- +rpRetryCount ∷ Lens' RetryPolicy Int +rpRetryCount = lens _rpRetryCount $ \rp n → rp { _rpRetryCount = n } + +-- | The default retry policy will attempt @5@ retries for a message. +-- +defaultRetryPolicy ∷ RetryPolicy +defaultRetryPolicy = RetryPolicy + { _rpRetryCount = 5 + } + type Message = T.Text data MessageQueueItem = MessageQueueItem { _mqiMessage ∷ !Message + -- ^ The contents of the message + , _mqiPartitionKey ∷ !Kin.PartitionKey + -- ^ The partition key the message is destined for + + , _mqiRemainingAttempts ∷ !Int + -- ^ The number of times remaining to try and publish this message } deriving (Eq, Show) mqiMessage ∷ Lens' MessageQueueItem Message @@ -153,6 +193,15 @@ mqiPartitionKey ∷ Lens' MessageQueueItem Kin.PartitionKey mqiPartitionKey = lens _mqiPartitionKey $ \i s → i { _mqiPartitionKey = s } +mqiRemainingAttempts ∷ Lens' MessageQueueItem Int +mqiRemainingAttempts = lens _mqiRemainingAttempts $ \i n → i { _mqiRemainingAttempts = n } + +messageQueueItemIsEligible + ∷ MessageQueueItem + → Bool +messageQueueItemIsEligible = + (≥ 1) ∘ _mqiRemainingAttempts + -- | The basic input required to construct a Kinesis producer. -- data ProducerKit @@ -166,6 +215,9 @@ , _pkBatchPolicy ∷ !BatchPolicy -- ^ The record batching policy for the producer. + , _pkRetryPolicy ∷ !RetryPolicy + -- ^ The retry policy for the producer. + , _pkMessageQueueBounds ∷ {-# UNPACK #-} !Int -- ^ The maximum number of records that may be enqueued at one time. @@ -188,6 +240,11 @@ pkBatchPolicy ∷ Lens' ProducerKit BatchPolicy pkBatchPolicy = lens _pkBatchPolicy $ \pk bp → pk { _pkBatchPolicy = bp } +-- | A lens for '_pkRetryPolicy'. +-- +pkRetryPolicy ∷ Lens' ProducerKit RetryPolicy +pkRetryPolicy = lens _pkRetryPolicy $ \pk rp → pk { _pkRetryPolicy = rp } + -- | A lens for '_pkMessageQueueBounds'. -- pkMessageQueueBounds ∷ Lens' ProducerKit Int @@ -200,14 +257,18 @@ -- | The (abstract) Kinesis producer client. -- -newtype KinesisProducer +data KinesisProducer = KinesisProducer - { _kpMessageQueue ∷ TBMQueue MessageQueueItem + { _kpMessageQueue ∷ !(TBMQueue MessageQueueItem) + , _kpRetryPolicy ∷ !RetryPolicy } kpMessageQueue ∷ Getter KinesisProducer (TBMQueue MessageQueueItem) kpMessageQueue = to _kpMessageQueue +kpRetryPolicy ∷ Getter KinesisProducer RetryPolicy +kpRetryPolicy = to _kpRetryPolicy + data ProducerError = KinesisError !SomeException -- ^ Represents an error which occured as a result of a request to Kinesis. @@ -216,6 +277,9 @@ -- ^ Thrown when a message could not be enqueued since the queue was full. -- This error must be handled at the call-site. + | MessageTooLarge + -- ^ Thrown when the message was larger than the maximum message size ('MaxMessageSize'). + | InvalidConcurrentConsumerCount -- ^ Thrown when 'pkMaxConcurrency' is set with an invalid value. @@ -239,6 +303,14 @@ MessageNotEnqueued m → Right m e → Left e +-- | A prism for 'MessageTooLarge'. +-- +_MessageTooLarge ∷ Prism' ProducerError () +_MessageTooLarge = + prism (const MessageTooLarge) $ \case + MessageTooLarge → Right () + e → Left e + -- | A prism for 'InvalidConcurrentConsumerCount'. -- _InvalidConcurrentConsumerCount ∷ Prism' ProducerError () @@ -375,17 +447,18 @@ putStrLn $ "Error: " ++ show e putStrLn "Will wait 5s" threadDelay 5000000 - leftover item + leftover $ item & mqiRemainingAttempts -~ 1 - handleError handler $ do - let partitionKey = item ^. mqiPartitionKey - void ∘ lift ∘ liftKinesis $ runKinesis Kin.PutRecord - { Kin.putRecordData = item ^. mqiMessage ∘ to T.encodeUtf8 - , Kin.putRecordExplicitHashKey = Nothing - , Kin.putRecordPartitionKey = partitionKey - , Kin.putRecordSequenceNumberForOrdering = Nothing - , Kin.putRecordStreamName = streamName - } + when (messageQueueItemIsEligible item) $ + handleError handler $ do + let partitionKey = item ^. mqiPartitionKey + void ∘ lift ∘ liftKinesis $ runKinesis Kin.PutRecord + { Kin.putRecordData = item ^. mqiMessage ∘ to T.encodeUtf8 + , Kin.putRecordExplicitHashKey = Nothing + , Kin.putRecordPartitionKey = partitionKey + , Kin.putRecordSequenceNumberForOrdering = Nothing + , Kin.putRecordStreamName = streamName + } splitEvery ∷ Int @@ -408,31 +481,34 @@ maxWorkerCount ← view pkMaxConcurrency awaitForever $ \messages → do let batches = splitEvery batchSize messages - leftovers ← lift ∘ flip (mapConcurrentlyN maxWorkerCount 100) batches $ \ms → do - let handler e = do - liftIO $ print e - return ms - - handleError handler $ do - items ← for ms $ \m → do - let partitionKey = m ^. mqiPartitionKey - return Kin.PutRecordsRequestEntry - { Kin.putRecordsRequestEntryData = m ^. mqiMessage ∘ to T.encodeUtf8 - , Kin.putRecordsRequestEntryExplicitHashKey = Nothing - , Kin.putRecordsRequestEntryPartitionKey = partitionKey - } + leftovers ← lift ∘ flip (mapConcurrentlyN maxWorkerCount 100) batches $ \items → do + case filter messageQueueItemIsEligible items of + [] → return [] + eligibleItems → do + handleError (\e → eligibleItems <$ liftIO (print e)) $ do + requestEntries ← for eligibleItems $ \m → do + let partitionKey = m ^. mqiPartitionKey + return Kin.PutRecordsRequestEntry + { Kin.putRecordsRequestEntryData = m ^. mqiMessage ∘ to T.encodeUtf8 + , Kin.putRecordsRequestEntryExplicitHashKey = Nothing + , Kin.putRecordsRequestEntryPartitionKey = partitionKey + } - Kin.PutRecordsResponse{..} ← liftKinesis $ runKinesis Kin.PutRecords - { Kin.putRecordsRecords = items - , Kin.putRecordsStreamName = streamName - } - let processResult m m' - | isJust (Kin.putRecordsResponseRecordErrorCode m') = Just m - | otherwise = Nothing - return ∘ catMaybes $ zipWith processResult ms putRecordsResponseRecords + Kin.PutRecordsResponse{..} ← liftKinesis $ runKinesis Kin.PutRecords + { Kin.putRecordsRecords = requestEntries + , Kin.putRecordsStreamName = streamName + } + let + processResult m m' + | isJust (Kin.putRecordsResponseRecordErrorCode m') = Just m + | otherwise = Nothing + return ∘ catMaybes $ zipWith processResult eligibleItems putRecordsResponseRecords - forM_ leftovers $ \mss → - unless (null mss) $ leftover mss + forM_ leftovers $ \items → + unless (null items) $ + leftover $ items + <&> mqiRemainingAttempts -~ 1 + & filter messageQueueItemIsEligible sendMessagesSink ∷ MonadProducerInternal m @@ -453,11 +529,15 @@ → Message → m () writeProducer producer !msg = do + when (T.length msg > MaxMessageSize) $ + throwError MessageTooLarge + gen ← liftIO R.newStdGen result ← liftIO ∘ atomically $ do tryWriteTBMQueue (producer ^. kpMessageQueue) MessageQueueItem { _mqiMessage = msg , _mqiPartitionKey = generatePartitionKey gen + , _mqiRemainingAttempts = producer ^. kpRetryPolicy . rpRetryCount . to succ } case result of Just True → return () @@ -516,7 +596,10 @@ Codensity $ \inner → do link consumerHandle - res ← inner $ KinesisProducer messageQueue + res ← inner KinesisProducer + { _kpMessageQueue = messageQueue + , _kpRetryPolicy = kit ^. pkRetryPolicy + } () ← wait consumerHandle return res