hw-kafka-client 2.3.0 → 2.3.1

raw patch · 4 files changed

+28/−14 lines, 4 files

Files

• example/ConsumerExample.hs +4/−0
• hw-kafka-client.cabal +1/−1
• src/Kafka/Consumer/Callbacks.hs +15/−12
• src/Kafka/Consumer/Types.hs +8/−1

@@ -41,8 +41,12 @@  printingRebalanceCallback :: KafkaConsumer -> RebalanceEvent -> IO () printingRebalanceCallback _ e = case e of+    RebalanceBeforeAssign ps ->+        putStrLn $ "[Rebalance] About to assign partitions: " <> show ps     RebalanceAssign ps ->         putStrLn $ "[Rebalance] Assign partitions: " <> show ps+    RebalanceBeforeRevoke ps ->+        putStrLn $ "[Rebalance] About to revoke partitions: " <> show ps     RebalanceRevoke ps ->         putStrLn $ "[Rebalance] Revoke partitions: " <> show ps 

@@ -1,5 +1,5 @@ name:                hw-kafka-client-version:             2.3.0+version:             2.3.1 homepage:            https://github.com/haskell-works/hw-kafka-client bug-reports:         https://github.com/haskell-works/hw-kafka-client/issues license:             MIT

@@ -69,18 +69,21 @@                           -> KafkaError                           -> [TopicPartition] -> IO () setRebalanceCallback f k e ps =-    case e of-        KafkaResponseError RdKafkaRespErrAssignPartitions -> do-            mbq <- getRdMsgQueue $ getKafkaConf k-            case mbq of-              Nothing -> pure ()-              Just mq -> forM_ ps (\tp -> redirectPartitionQueue (getKafka k) (tpTopicName tp) (tpPartition tp) mq)-            void $ assign k ps-            f k (RebalanceAssign ((tpTopicName &&& tpPartition) <$> ps))-        KafkaResponseError RdKafkaRespErrRevokePartitions -> do-            void $ assign k []-            f k (RebalanceRevoke ((tpTopicName &&& tpPartition) <$> ps))-        x -> error $ "Rebalance: UNKNOWN response: " <> show x+  let assignment = (tpTopicName &&& tpPartition) <$> ps+  in case e of+    KafkaResponseError RdKafkaRespErrAssignPartitions -> do+        mbq <- getRdMsgQueue $ getKafkaConf k+        case mbq of+          Nothing -> pure ()+          Just mq -> forM_ ps (\tp -> redirectPartitionQueue (getKafka k) (tpTopicName tp) (tpPartition tp) mq)+        f k (RebalanceBeforeAssign assignment)+        void $ assign k ps+        f k (RebalanceAssign assignment)+    KafkaResponseError RdKafkaRespErrRevokePartitions -> do+        f k (RebalanceBeforeRevoke assignment)+        void $ assign k []+        f k (RebalanceRevoke assignment)+    x -> error $ "Rebalance: UNKNOWN response: " <> show x  -- | Assigns specified partitions to a current consumer. -- Assigning an empty list means unassigning from all partitions that are currently assigned.

@@ -28,8 +28,15 @@ newtype Offset          = Offset Int64 deriving (Show, Eq, Read) data OffsetReset        = Earliest | Latest deriving (Show, Eq) +-- | A set of events which happen during the rebalancing process data RebalanceEvent =-    RebalanceAssign [(TopicName, PartitionId)]+    -- | Happens before Kafka Client confirms new assignment+    RebalanceBeforeAssign [(TopicName, PartitionId)]+    -- | Happens after the new assignment is confirmed+  | RebalanceAssign [(TopicName, PartitionId)]+    -- | Happens before Kafka Client confirms partitions rejection+  | RebalanceBeforeRevoke [(TopicName, PartitionId)]+    -- | Happens after the rejection is confirmed   | RebalanceRevoke [(TopicName, PartitionId)]   deriving (Eq, Show)