milena 0.3.0.0 → 0.4.0.0

raw patch · 6 files changed

+273/−101 lines, 6 filesdep +pretty-showdep +semigroupsdep +tastydep −eitherdep −hspecdep ~basedep ~bytestringdep ~cerealnew-component:exe:exkhsPVP ok

version bump matches the API change (PVP)

Dependencies added: pretty-show, semigroups, tasty, tasty-hspec, tasty-quickcheck

Dependencies removed: either, hspec

Dependency ranges changed: base, bytestring, cereal, lens, lifted-base, mtl

API changes (from Hackage documentation)

Files

• Network/Kafka.hs +152/−62
• Network/Kafka/Producer.hs +6/−27
• Network/Kafka/Protocol.hs +1/−1
• milena.cabal +23/−6
• src/Main.hs +42/−0
• test/tests.hs +49/−5

@@ -5,13 +5,16 @@ module Network.Kafka where  import Control.Applicative-import Control.Exception (bracket)+import Control.Exception (IOException)+import Control.Exception.Lifted (catch) import Control.Lens import Control.Monad (liftM)+import Control.Monad.Except (ExceptT(..), runExceptT, withExceptT, MonadError(..)) import Control.Monad.Trans (liftIO, lift)-import Control.Monad.Trans.Either import Control.Monad.Trans.State import Data.ByteString.Char8 (ByteString)+import Data.List.NonEmpty (NonEmpty(..))+import qualified Data.List.NonEmpty as NE import Data.Monoid ((<>)) import qualified Data.Pool as Pool import Data.Serialize.Get@@ -19,9 +22,12 @@ import qualified Data.ByteString.Char8 as B import qualified Data.Map as M import qualified Network+import Prelude  import Network.Kafka.Protocol +type KafkaAddress = (Host, Port)+ data KafkaState = KafkaState { -- | Name to use as a client ID.                                _stateName :: KafkaString                                -- | How many acknowledgements are required for producing.@@ -39,23 +45,18 @@                                -- | Broker cache                              , _stateBrokers :: M.Map Leader Broker                                -- | Connection cache-                             , _stateConnections :: M.Map Broker (Pool.Pool Handle)+                             , _stateConnections :: M.Map KafkaAddress (Pool.Pool Handle)                                -- | Topic metadata cache                              , _stateTopicMetadata :: M.Map TopicName TopicMetadata-                             }+                               -- | Address cache+                             , _stateAddresses :: NonEmpty KafkaAddress+                             } deriving (Show)  makeLenses ''KafkaState -data KafkaClient = KafkaClient { _kafkaClientState :: KafkaState-                               , _kafkaClientHandle :: Handle-                               }--makeLenses ''KafkaClient- -- | The core Kafka monad.-type Kafka = StateT KafkaClient (EitherT KafkaClientError IO)+type Kafka = StateT KafkaState (ExceptT KafkaClientError IO) -type KafkaAddress = (Host, Port) type KafkaClientId = KafkaString  -- | Errors given from the Kafka monad.@@ -68,6 +69,7 @@                         -- | Could not find a cached broker for the found leader.                       | KafkaInvalidBroker Leader                       | KafkaFailedToFetchMetadata+                      | KafkaIOException IOException                         deriving (Eq, Show)  -- | Type of response to expect, used for 'KafkaExpected' error.@@ -136,8 +138,8 @@ defaultMaxWaitTime = 0  -- | Create a consumer using default values.-defaultState :: KafkaClientId -> KafkaState-defaultState cid =+mkKafkaState :: KafkaClientId -> KafkaAddress -> KafkaState+mkKafkaState cid addy =     KafkaState cid                defaultRequiredAcks                defaultRequestTimeout@@ -148,50 +150,89 @@                M.empty                M.empty                M.empty+               (addy :| []) --- | Run the underlying Kafka monad at the given leader address and initial state.-runKafka :: KafkaAddress -> KafkaState -> Kafka a -> IO (Either KafkaClientError a)-runKafka (h, p) s k =-  bracket (Network.connectTo (h ^. hostString) (p ^. portId)) hClose $ runEitherT . evalStateT k . KafkaClient s+addKafkaAddress :: KafkaAddress -> KafkaState -> KafkaState+addKafkaAddress = over stateAddresses . NE.nub .: cons+  where infixr 9 .:+        (.:) :: (c -> d) -> (a -> b -> c) -> a -> b -> d+        (.:) = (.).(.) +-- | Run the underlying Kafka monad.+runKafka :: KafkaState -> Kafka a -> IO (Either KafkaClientError a)+runKafka s k = runExceptT $ evalStateT k s+ -- | Make a request, incrementing the `_stateCorrelationId`. makeRequest :: RequestMessage -> Kafka Request makeRequest m = do-  corid <- use (kafkaClientState . stateCorrelationId)-  kafkaClientState . stateCorrelationId += 1-  conid <- use (kafkaClientState . stateName)+  corid <- use stateCorrelationId+  stateCorrelationId += 1+  conid <- use stateName   return $ Request (corid, ClientId conid, m) --- | Perform a request and deserialize the response.-doRequest :: Request -> Kafka Response-doRequest r = do-  h <- use kafkaClientHandle-  doRequest' h r+-- | Catch 'IOException's and wrap them in 'KafkaIOException's.+tryKafkaIO :: IO a -> Kafka a+tryKafkaIO = tryKafka . liftIO -doRequest' :: Handle -> Request -> Kafka Response-doRequest' h r = mapStateT (bimapEitherT KafkaDeserializationError id) $ do-  dataLength <- lift . EitherT $ do+-- | Catch 'IOException's and wrap them in 'KafkaIOException's.+tryKafka :: Kafka a -> Kafka a+tryKafka = (`catch` \e -> lift . throwError $ KafkaIOException (e :: IOException))++doRequest :: Handle -> Request -> Kafka Response+doRequest h r = do+  rawLength <- tryKafkaIO $ do     B.hPut h $ requestBytes r     hFlush h-    rawLength <- B.hGet h 4-    return $ runGet (liftM fromIntegral getWord32be) rawLength-  resp <- liftIO $ B.hGet h dataLength-  lift . hoistEither $ runGet (getResponse dataLength) resp+    B.hGet h 4+  dataLength <- runGetKafka (liftM fromIntegral getWord32be) rawLength+  resp <- tryKafkaIO $ B.hGet h dataLength+  runGetKafka (getResponse dataLength) resp +runGetKafka :: Get a -> ByteString -> Kafka a+runGetKafka g bs = lift $ withExceptT KafkaDeserializationError $ ExceptT $ return $ runGet g bs+ -- | Send a metadata request metadata :: MetadataRequest -> Kafka MetadataResponse-metadata request = do-  h <- use kafkaClientHandle-  metadata' h request+metadata request = withAnyHandle $ flip metadata' request  -- | Send a metadata request metadata' :: Handle -> MetadataRequest -> Kafka MetadataResponse-metadata' handle request =-    makeRequest (MetadataRequest request) >>= doRequest' handle >>= expectResponse ExpectedMetadata _MetadataResponse+metadata' h request =+    makeRequest (MetadataRequest request) >>= doRequest h >>= expectResponse ExpectedMetadata _MetadataResponse +getTopicPartitionLeader :: TopicName -> Partition -> Kafka Broker+getTopicPartitionLeader t p = do+  let s = stateTopicMetadata . at t+  tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata+  leader <- expect KafkaFailedToFetchMetadata (firstOf $ findPartitionMetadata t . (folded . findPartition p) . partitionMetadataLeader) tmd+  use stateBrokers >>= expect (KafkaInvalidBroker leader) (view $ at leader)++expect :: KafkaClientError -> (a -> Maybe b) -> a -> Kafka b+expect e f = lift . maybe (throwError e) return . f++-- | Find a leader and partition for the topic.+brokerPartitionInfo :: TopicName -> Kafka [PartitionAndLeader]+brokerPartitionInfo t = do+  let s = stateTopicMetadata . at t+  tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata+  return $ pal <$> tmd ^. partitionsMetadata+    where pal d = PartitionAndLeader t (d ^. partitionId) (d ^. partitionMetadataLeader)++findMetadataOrElse :: [TopicName] -> Getting (Maybe a) KafkaState (Maybe a) -> KafkaClientError -> Kafka a+findMetadataOrElse ts s err = do+  maybeFound <- use s+  case maybeFound of+    Just x -> return x+    Nothing -> do+      updateMetadatas ts+      maybeFound' <- use s+      case maybeFound' of+        Just x -> return x+        Nothing -> lift $ throwError err+ -- | Function to give an error when the response seems wrong. expectResponse :: KafkaExpectedResponse -> Getting (Leftmost b) ResponseMessage b -> Response -> Kafka b-expectResponse e p = lift . maybe (left $ KafkaExpected e) return . firstOf (responseMessage . p)+expectResponse e p = expect (KafkaExpected e) (firstOf $ responseMessage . p)  -- | Convert an abstract time to a serializable protocol value. protocolTime :: KafkaTime -> Time@@ -208,15 +249,20 @@ -- | Construct a fetch request from the values in the state. fetchRequest :: Offset -> Partition -> TopicName -> Kafka FetchRequest fetchRequest o p topic = do-  wt <- use (kafkaClientState . stateWaitTime)-  ws <- use (kafkaClientState . stateWaitSize)-  bs <- use (kafkaClientState . stateBufferSize)+  wt <- use stateWaitTime+  ws <- use stateWaitSize+  bs <- use stateBufferSize   return $ FetchReq (ordinaryConsumerId, wt, ws, [(topic, [(p, o, bs)])])  -- | Execute a fetch request and get the raw fetch response.+fetch' :: Handle -> FetchRequest -> Kafka FetchResponse+fetch' h request =+    makeRequest (FetchRequest request) >>= doRequest h >>= expectResponse ExpectedFetch _FetchResponse++-- | Execute a fetch request and get the raw fetch response. Round-robins the+-- requests to addresses in the 'KafkaState'. fetch :: FetchRequest -> Kafka FetchResponse-fetch request =-    makeRequest (FetchRequest request) >>= doRequest >>= expectResponse ExpectedFetch _FetchResponse+fetch request = withAnyHandle $ flip fetch' request  -- | Extract out messages with their topics from a fetch response. fetchMessages :: FetchResponse -> [TopicAndMessage]@@ -227,8 +273,10 @@ updateMetadatas ts = do   md <- metadata $ MetadataReq ts   let (brokers, tmds) = (md ^.. metadataResponseBrokers . folded, md ^.. topicsMetadata . folded)-  kafkaClientState . stateBrokers %= \m -> foldr addBroker m brokers-  kafkaClientState . stateTopicMetadata %= \m -> foldr addTopicMetadata m tmds+      addresses = map broker2address brokers+  stateAddresses %= NE.nub . NE.fromList . (++ addresses) . NE.toList+  stateBrokers %= \m -> foldr addBroker m brokers+  stateTopicMetadata %= \m -> foldr addTopicMetadata m tmds   return ()     where addBroker :: Broker -> M.Map Leader Broker -> M.Map Leader Broker           addBroker b = M.insert (Leader . Just $ b ^. brokerNode . nodeId) b@@ -241,26 +289,63 @@ updateAllMetadata :: Kafka () updateAllMetadata = updateMetadatas [] --- | Execute a handler action, creating a new Pool and updating the connections Map if needed.+-- | Execute a Kafka action with a 'Handle' for the given 'Broker', updating+-- the connections cache if needed.+--+-- When the action throws an 'IOException', it is caught and returned as a+-- 'KafkaIOException' in the Kafka monad.+--+-- Note that when the given action throws an exception, any state changes will+-- be discarded. This includes both 'IOException's and exceptions thrown by+-- 'throwError' from 'Control.Monad.Except'. withBrokerHandle :: Broker -> (Handle -> Kafka a) -> Kafka a-withBrokerHandle broker f = do-  conns <- use (kafkaClientState . stateConnections)-  let foundPool = conns ^. at broker+withBrokerHandle broker = withAddressHandle (broker2address broker)++-- | Execute a Kafka action with a 'Handle' for the given 'KafkaAddress',+-- updating the connections cache if needed.+--+-- When the action throws an 'IOException', it is caught and returned as a+-- 'KafkaIOException' in the Kafka monad.+--+-- Note that when the given action throws an exception, any state changes will+-- be discarded. This includes both 'IOException's and exceptions thrown by+-- 'throwError' from 'Control.Monad.Except'.+withAddressHandle :: KafkaAddress -> (Handle -> Kafka a) -> Kafka a+withAddressHandle address kafkaAction = do+  conns <- use stateConnections+  let foundPool = conns ^. at address   pool <- case foundPool of     Nothing -> do-      newPool <- liftIO $ mkPool broker-      kafkaClientState . stateConnections .= (at broker ?~ newPool $ conns)+      newPool <- tryKafkaIO $ mkPool address+      stateConnections .= (at address ?~ newPool $ conns)       return newPool     Just p -> return p-  Pool.withResource pool f-    where mkPool :: Broker -> IO (Pool.Pool Handle)-          mkPool b = Pool.createPool (createHandle b) hClose 1 10 1-          createHandle b = do-            let h = b ^. brokerHost ^. hostString-                p = b ^. brokerPort ^. portId-            Network.connectTo h p+  tryKafka $ Pool.withResource pool kafkaAction+    where+      mkPool :: KafkaAddress -> IO (Pool.Pool Handle)+      mkPool a = Pool.createPool (createHandle a) hClose 1 10 1+        where createHandle (h, p) = Network.connectTo (h ^. hostString) (p ^. portId) +broker2address :: Broker -> KafkaAddress+broker2address broker = (,) (broker ^. brokerHost) (broker ^. brokerPort) +-- | Like 'withAddressHandle', but round-robins the addresses in the 'KafkaState'.+--+-- When the action throws an 'IOException', it is caught and returned as a+-- 'KafkaIOException' in the Kafka monad.+--+-- Note that when the given action throws an exception, any state changes will+-- be discarded. This includes both 'IOException's and exceptions thrown by+-- 'throwError' from 'Control.Monad.Except'.+withAnyHandle :: (Handle -> Kafka a) -> Kafka a+withAnyHandle f = do+  (addy :| _) <- use stateAddresses+  x <- withAddressHandle addy f+  stateAddresses %= rotate+  return x+    where rotate :: NonEmpty a -> NonEmpty a+          rotate = NE.fromList . rotate' 1 . NE.toList+          rotate' n xs = zipWith const (drop n (cycle xs)) xs  -- * Offsets @@ -272,11 +357,16 @@                                , _maxNumOffsets :: MaxNumberOfOffsets                                } --- TODO: Properly look up the offset via the partition. -- | Get the first found offset. getLastOffset :: KafkaTime -> Partition -> TopicName -> Kafka Offset-getLastOffset m p t =-    makeRequest (offsetRequest [(TopicAndPartition t p, PartitionOffsetRequestInfo m 1)]) >>= doRequest >>= maybe (StateT . const $ left KafkaNoOffset) return . firstOf (responseMessage . _OffsetResponse . offsetResponseOffset p)+getLastOffset m p t = do+  broker <- getTopicPartitionLeader t p+  withBrokerHandle broker (\h -> getLastOffset' h m p t)++-- | Get the first found offset.+getLastOffset' :: Handle -> KafkaTime -> Partition -> TopicName -> Kafka Offset+getLastOffset' h m p t =+    makeRequest (offsetRequest [(TopicAndPartition t p, PartitionOffsetRequestInfo m 1)]) >>= doRequest h >>= maybe (StateT . const $ throwError KafkaNoOffset) return . firstOf (responseMessage . _OffsetResponse . offsetResponseOffset p)  -- | Create an offset request. offsetRequest :: [(TopicAndPartition, PartitionOffsetRequestInfo)] -> RequestMessage

@@ -1,10 +1,8 @@ module Network.Kafka.Producer where -import Prelude hiding ((!!)) import Control.Applicative import Control.Lens-import Control.Monad.Trans (liftIO, lift)-import Control.Monad.Trans.Either+import Control.Monad.Trans (liftIO) import Data.ByteString.Char8 (ByteString) import qualified Data.Digest.Murmur32 as Murmur32 import Data.List.Safe ((!!))@@ -12,6 +10,7 @@ import System.IO import qualified Data.Map as M import System.Random (getStdRandom, randomR)+import Prelude hiding ((!!))  import Network.Kafka import Network.Kafka.Protocol@@ -21,7 +20,7 @@ -- | Execute a produce request and get the raw preduce response. produce :: Handle -> ProduceRequest -> Kafka ProduceResponse produce handle request =-    makeRequest (ProduceRequest request) >>= doRequest' handle >>= expectResponse ExpectedProduce _ProduceResponse+    makeRequest (ProduceRequest request) >>= doRequest handle >>= expectResponse ExpectedProduce _ProduceResponse  -- | Construct a produce request with explicit arguments. produceRequest :: RequiredAcks -> Timeout -> [(TopicAndPartition, MessageSet)] -> ProduceRequest@@ -65,32 +64,12 @@ -- | Execute a produce request using the values in the state. send :: Leader -> [(TopicAndPartition, MessageSet)] -> Kafka ProduceResponse send l ts = do-  let s = kafkaClientState . stateBrokers . at l+  let s = stateBrokers . at l       topicNames = map (_tapTopic . fst) ts   broker <- findMetadataOrElse topicNames s (KafkaInvalidBroker l)-  requiredAcks <- use (kafkaClientState . stateRequiredAcks)-  requestTimeout <- use (kafkaClientState . stateRequestTimeout)+  requiredAcks <- use stateRequiredAcks+  requestTimeout <- use stateRequestTimeout   withBrokerHandle broker $ \handle -> produce handle $ produceRequest requiredAcks requestTimeout ts---- | Find a leader and partition for the topic.-brokerPartitionInfo :: TopicName -> Kafka [PartitionAndLeader]-brokerPartitionInfo t = do-  let s = kafkaClientState . stateTopicMetadata . at t-  tmd <- findMetadataOrElse [t] s KafkaFailedToFetchMetadata-  return $ pal <$> tmd ^. partitionsMetadata-    where pal d = PartitionAndLeader t (d ^. partitionId) (d ^. partitionMetadataLeader)--findMetadataOrElse :: [TopicName] -> Getting (Maybe a) KafkaClient (Maybe a) -> KafkaClientError -> Kafka a-findMetadataOrElse ts s err = do-  maybeFound <- use s-  case maybeFound of-    Just x -> return x-    Nothing -> do-      updateMetadatas ts-      maybeFound' <- use s-      case maybeFound' of-        Just x -> return x-        Nothing -> lift $ left $ err  getRandPartition :: [PartitionAndLeader] -> Kafka (Maybe PartitionAndLeader) getRandPartition ps =

@@ -5,7 +5,7 @@  module Network.Kafka.Protocol where -import Control.Applicative (Applicative(..), Alternative(..), (<$>), (<*>))+import Control.Applicative import Control.Category (Category(..)) import Control.Lens import Control.Monad (replicateM, liftM, liftM2, liftM3, liftM4, liftM5)

@@ -4,7 +4,7 @@ -- PVP summary:      +-+------- breaking API changes --                   | | +----- non-breaking API additions --                   | | | +--- code changes with no API change-version:             0.3.0.0+version:             0.4.0.0 synopsis:            A Kafka client for Haskell. description:     The protocol module is stable (the only changes will be to support changes in the Kafka protocol). The API is functional but subject to change.@@ -25,13 +25,13 @@ source-repository this   type: git   location: https://github.com/tylerholien/milena.git-  tag: 0.3.0.0+  tag: 0.4.0.0  library   default-language: Haskell2010   ghc-options:         -Wall   exposed-modules:     Network.Kafka,-                       Network.Kafka.Protocol+                       Network.Kafka.Protocol,                        Network.Kafka.Producer   build-depends:       base          >=4.7      && <5,                        mtl           >=2.1      && <2.3,@@ -40,14 +40,14 @@                        network       >=2.4      && <2.7,                        digest        >=0.0.1.0  && <0.1,                        containers    >=0.5      && <0.6,-                       either        >=4.3      && <4.4,                        random        >=1.0      && <1.2,                        transformers  >=0.3      && <0.5,                        lens          >=4.4      && <4.13,                        resource-pool >=0.2.3.2  && <0.3,                        lifted-base   >=0.2.3.6  && <0.3,                        murmur-hash   >=0.1.0.8  && <0.2,-                       listsafe      >=0.1.0.1  && <0.2+                       listsafe      >=0.1.0.1  && <0.2,+                       semigroups    >=0.16.2.2 && <0.17  test-suite test   default-language: Haskell2010@@ -58,7 +58,24 @@                        network,                        QuickCheck,                        bytestring,-                       hspec+                       lens,+                       semigroups,+                       tasty,+                       tasty-hspec,+                       tasty-quickcheck   hs-source-dirs:      test   main-is:             tests.hs   type:                exitcode-stdio-1.0++executable exkhs+  hs-source-dirs:      src+  main-is:             Main.hs+  default-language:    Haskell2010+  build-depends:       base,+                       milena,+                       bytestring,+                       cereal,+                       mtl,+                       lifted-base,+                       lens,+                       pretty-show

@@ -0,0 +1,42 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE UnicodeSyntax #-}+{-# LANGUAGE ExplicitForAll #-}++module Main where++import qualified Data.ByteString.Char8 as C+import Control.Lens+import Control.Monad (forM)+import Control.Monad.Except (catchError)+import Network.Kafka+import Network.Kafka.Protocol+import System.Environment (getArgs)+-- import Text.Show.Pretty (ppShow)++main ∷ IO ()+main = do+  h:portString:_ ← getArgs+  let host  = Host (KString (C.pack h))+      port  = Port (read portString)+      topic = "open_channel"+      state = mkKafkaState "command-line-test-client" (host, port)+  result ← runKafka state $ do+    -- md ← metadata (MetadataReq [topic])+    -- putStrLnM (ppShow md)+    topicPartitionList ← brokerPartitionInfo topic+    forM topicPartitionList $ \(PartitionAndLeader { _palLeader, _palTopic, _palPartition }) → do+      let s = stateBrokers . at _palLeader+      broker ← findMetadataOrElse [topic] s (KafkaInvalidBroker _palLeader)+      flip catchError (return . Left) $ do+        result ← withBrokerHandle broker $ \handle → do+          offset ← getLastOffset' handle EarliestTime _palPartition topic+          fetchRequest offset _palPartition topic >>= fetch' handle+        return $ Right result+  print result++-- tmd :: TopicMetadata+-- tmd = TopicMetadata (NoError, "omfg", [PartitionMetadata (NoError, 0, Leader (Just 1), Replicas [], Isr []), PartitionMetadata (NoError, 1, Leader (Just 1), Replicas [], Isr [])])+-- newtype TopicMetadata = TopicMetadata { _topicMetadataFields :: (KafkaError, TopicName, [PartitionMetadata]) } deriving (Show, Eq, Deserializable)+-- newtype PartitionMetadata = PartitionMetadata { _partitionMetadataFields :: (KafkaError, Partition, Leader, Replicas, Isr) } deriving (Show, Eq, Deserializable)

@@ -3,18 +3,28 @@ module Main where  import Data.Functor-import Data.Either (isRight)+import Data.Either (isRight, isLeft)+import qualified Data.List.NonEmpty as NE+import Control.Lens+import Control.Monad.Except (catchError, throwError)+import Control.Monad.Trans (liftIO) import Network.Kafka import Network.Kafka.Producer import Network.Kafka.Protocol (Leader(..))-import Test.Hspec-import Test.Hspec.QuickCheck+import Test.Tasty+import Test.Tasty.Hspec+import Test.Tasty.QuickCheck import qualified Data.ByteString.Char8 as B +import Prelude+ main :: IO ()-main = hspec $ do+main = testSpec "the specs" specs >>= defaultMain++specs :: Spec+specs = do   let topic = "milena-test"-      run = runKafka ("localhost", 9092) $ defaultState "milena-test-client"+      run = runKafka $ mkKafkaState "milena-test-client" ("localhost", 9092)       byteMessages = fmap (TopicAndMessage topic . makeMessage . B.pack)    describe "can talk to local Kafka server" $ do@@ -22,6 +32,13 @@       result <- run . produceMessages $ byteMessages ms       result `shouldSatisfy` isRight +    prop "can produce multiple messages" $ \(ms, ms') -> do+      result <- run $ do+        r1 <- produceMessages $ byteMessages ms+        r2 <- produceMessages $ byteMessages ms'+        return $ r1 ++ r2+      result `shouldSatisfy` isRight+     prop "can fetch messages" $ do       result <- run $ do         offset <- getLastOffset EarliestTime 0 topic@@ -37,3 +54,30 @@         void $ send leader [(TopicAndPartition topic 0, groupMessagesToSet messages)]         fmap tamPayload . fetchMessages <$> (fetch =<< fetchRequest offset 0 topic)       result `shouldBe` Right (tamPayload <$> messages)++  describe "withAddressHandle" $ do+    it "turns 'IOException's into 'KafkaClientError's" $ do+      result <- run $ withAddressHandle ("localhost", 9092) (\_ -> liftIO $ ioError $ userError "SOMETHING WENT WRONG!") :: IO (Either KafkaClientError ())+      result `shouldSatisfy` isLeft++    it "discards monadic effects when exceptions are thrown" $ do+      result <- run $ do+        stateName .= "expected"+        _ <- flip catchError (return . Left) $ withAddressHandle ("localhost", 9092) $ \_ -> do+          stateName .= "changed"+          _ <- throwError KafkaFailedToFetchMetadata+          n <- use stateName+          return (Right n)+        use stateName+      result `shouldBe` Right "expected"++  describe "updateMetadatas" $+    it "de-dupes _stateAddresses" $ do+      result <- run $ do+        stateAddresses %= NE.cons ("localhost", 9092)+        updateMetadatas []+        use stateAddresses+      result `shouldBe` fmap NE.nub result++prop :: Testable prop => String -> prop -> SpecWith ()+prop s = it s . property