diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,13 @@ +0.5.4.0 + +* commitOffset and fetchOffset to the API (https://github.com/adamflott/milena/pull/7) +* Add Delete Topic support +* Make tests pass on multiple invocations + +0.5.3.0 + +* Add Create Topic support (https://github.com/adamflott/milena/pull/6). Thanks matth! + 0.5.2.3 * Bump network upper bounds diff --git a/Network/Kafka.hs b/Network/Kafka.hs --- a/Network/Kafka.hs +++ b/Network/Kafka.hs @@ -202,6 +202,44 @@ CreateTopicsReq ([(topic, partition, replication_factor, replica_assignment, config)], defaultRequestTimeout) +deleteTopic :: Kafka m => DeleteTopicsRequest -> m DeleteTopicsResponse +deleteTopic request = withAnyHandle $ flip deleteTopic' request + +deleteTopic' :: Kafka m => Handle -> DeleteTopicsRequest -> m DeleteTopicsResponse +deleteTopic' h request = makeRequest h $ DeleteTopicsRR request + +deleteTopicsRequest :: TopicName -> DeleteTopicsRequest +deleteTopicsRequest topic = DeleteTopicsReq ([topic], defaultRequestTimeout) + + +fetchOffset :: Kafka m => OffsetFetchRequest -> m OffsetFetchResponse +fetchOffset request = withAnyHandle $ flip fetchOffset' request + +fetchOffset' :: Kafka m => Handle -> OffsetFetchRequest -> m OffsetFetchResponse +fetchOffset' h request = makeRequest h $ OffsetFetchRR request + +fetchOffsetRequest :: + ConsumerGroup -> TopicName -> Partition -> OffsetFetchRequest +fetchOffsetRequest consumerGroup topic partition = + OffsetFetchReq + (consumerGroup, [(topic, [partition])]) + + +commitOffset :: Kafka m => OffsetCommitRequest -> m OffsetCommitResponse +commitOffset request = withAnyHandle $ flip commitOffset' request + +commitOffset' :: + Kafka m => Handle -> OffsetCommitRequest -> m OffsetCommitResponse +commitOffset' h request = makeRequest h $ OffsetCommitRR request + +commitOffsetRequest :: + ConsumerGroup -> TopicName -> Partition -> Offset -> OffsetCommitRequest +commitOffsetRequest consumerGroup topic partition offset = + let time = -1 + metadata_ = Metadata "milena" + in OffsetCommitReq + (consumerGroup, [(topic, [(partition, offset, time, metadata_)])]) + getTopicPartitionLeader :: Kafka m => TopicName -> Partition -> m Broker getTopicPartitionLeader t p = do diff --git a/Network/Kafka/Protocol.hs b/Network/Kafka/Protocol.hs --- a/Network/Kafka/Protocol.hs +++ b/Network/Kafka/Protocol.hs @@ -2,24 +2,28 @@ ( module Network.Kafka.Protocol ) where +import Prelude hiding ((.), id) + +-- base import Control.Applicative import Control.Category (Category(..)) import Control.Exception (Exception) -import Control.Lens import Control.Monad (replicateM, liftM2, liftM3, liftM4, liftM5, unless) -import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Bits ((.&.)) +import Data.Int +import GHC.Exts (IsString(..)) +import GHC.Generics (Generic) +import System.IO + +-- Hackage +import Control.Lens +import Control.Monad.IO.Class (MonadIO, liftIO) import Data.ByteString.Char8 (ByteString) import Data.ByteString.Lens (unpackedChars) import Data.Digest.CRC32 -import Data.Int import Data.Serialize.Get import Data.Serialize.Put -import GHC.Exts (IsString(..)) -import GHC.Generics (Generic) -import System.IO import Numeric.Lens -import Prelude hiding ((.), id) import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy.Char8 as LB (fromStrict, toStrict) import qualified Codec.Compression.GZip as GZip (compress, decompress) @@ -31,6 +35,9 @@ FetchRR :: MonadIO m => FetchRequest -> ReqResp (m FetchResponse) OffsetRR :: MonadIO m => OffsetRequest -> ReqResp (m OffsetResponse) TopicsRR :: MonadIO m => CreateTopicsRequest -> ReqResp (m CreateTopicsResponse) + DeleteTopicsRR :: MonadIO m => DeleteTopicsRequest -> ReqResp (m DeleteTopicsResponse) + OffsetCommitRR :: MonadIO m => OffsetCommitRequest -> ReqResp (m OffsetCommitResponse) + OffsetFetchRR :: MonadIO m => OffsetFetchRequest -> ReqResp (m OffsetFetchResponse) doRequest' :: (Deserializable a, MonadIO m) => CorrelationId -> Handle -> Request -> m (Either String a) doRequest' correlationId h r = do @@ -53,6 +60,9 @@ doRequest clientId correlationId h (FetchRR req) = doRequest' correlationId h $ Request (correlationId, clientId, FetchRequest req) doRequest clientId correlationId h (OffsetRR req) = doRequest' correlationId h $ Request (correlationId, clientId, OffsetRequest req) doRequest clientId correlationId h (TopicsRR req) = doRequest' correlationId h $ Request (correlationId, clientId, CreateTopicsRequest req) +doRequest clientId correlationId h (DeleteTopicsRR req) = doRequest' correlationId h $ Request (correlationId, clientId, DeleteTopicsRequest req) +doRequest clientId correlationId h (OffsetCommitRR req) = doRequest' correlationId h $ Request (correlationId, clientId, OffsetCommitRequest req) +doRequest clientId correlationId h (OffsetFetchRR req) = doRequest' correlationId h $ Request (correlationId, clientId, OffsetFetchRequest req) class Serializable a where serialize :: a -> Put @@ -75,6 +85,7 @@ | OffsetFetchRequest OffsetFetchRequest | GroupCoordinatorRequest GroupCoordinatorRequest | CreateTopicsRequest CreateTopicsRequest + | DeleteTopicsRequest DeleteTopicsRequest deriving (Show, Generic, Eq) newtype MetadataRequest = MetadataReq [TopicName] deriving (Show, Eq, Serializable, Generic, Deserializable) @@ -106,6 +117,10 @@ TopicsResp { _topicsResponseFields :: [(TopicName, KafkaError)] } deriving (Show, Eq, Deserializable, Serializable, Generic) +newtype DeleteTopicsResponse = + DeleteTopicsResp { _deleteTopicsResponseFields :: [(TopicName, KafkaError)] } + deriving (Show, Eq, Deserializable, Serializable, Generic) + newtype MetadataResponse = MetadataResp { _metadataResponseFields :: ([Broker], [TopicMetadata]) } deriving (Show, Eq, Deserializable, Generic) newtype Broker = Broker { _brokerFields :: (NodeId, Host, Port) } deriving (Show, Eq, Ord, Deserializable, Generic) newtype NodeId = NodeId { _nodeId :: Int32 } deriving (Show, Eq, Deserializable, Num, Integral, Ord, Real, Enum, Generic) @@ -181,6 +196,8 @@ newtype GroupCoordinatorRequest = GroupCoordinatorReq ConsumerGroup deriving (Show, Eq, Serializable, Generic) newtype CreateTopicsRequest = CreateTopicsReq ([(TopicName, Partition, ReplicationFactor, [(Partition, Replicas)], [(KafkaString, Metadata)])], Timeout) deriving (Show, Eq, Serializable, Generic) +newtype DeleteTopicsRequest = DeleteTopicsReq ([TopicName], Timeout) deriving (Show, Eq, Serializable, Generic) + newtype OffsetCommitRequest = OffsetCommitReq (ConsumerGroup, [(TopicName, [(Partition, Offset, Time, Metadata)])]) deriving (Show, Eq, Serializable, Generic) newtype OffsetFetchRequest = OffsetFetchReq (ConsumerGroup, [(TopicName, [Partition])]) deriving (Show, Eq, Serializable, Generic) newtype ConsumerGroup = ConsumerGroup KafkaString deriving (Show, Eq, Serializable, Deserializable, IsString, Generic) @@ -286,6 +303,7 @@ apiKey OffsetFetchRequest{} = ApiKey 9 apiKey GroupCoordinatorRequest{} = ApiKey 10 apiKey CreateTopicsRequest{} = ApiKey 19 +apiKey DeleteTopicsRequest{} = ApiKey 20 instance Serializable RequestMessage where serialize (ProduceRequest r) = serialize r @@ -296,6 +314,7 @@ serialize (OffsetFetchRequest r) = serialize r serialize (GroupCoordinatorRequest r) = serialize r serialize (CreateTopicsRequest r) = serialize r + serialize (DeleteTopicsRequest r) = serialize r instance Serializable Int64 where serialize = putWord64be . fromIntegral instance Serializable Int32 where serialize = putWord32be . fromIntegral diff --git a/milena.cabal b/milena.cabal --- a/milena.cabal +++ b/milena.cabal @@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.1. +-- This file has been generated from package.yaml by hpack version 0.31.2. -- -- see: https://github.com/sol/hpack -- --- hash: 8060a8b3f6bf4aad48a041ce65dd55877fe0f1bd5353f32359bae70b010f2843 +-- hash: a7bdec96a93aec94bc09e9d89b957ff87d484a0d2f73237f8816de0a56cd03b4 name: milena -version: 0.5.3.0 +version: 0.5.4.0 synopsis: A Kafka client for Haskell. description: A Kafka client for Haskell. 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. diff --git a/test/tests.hs b/test/tests.hs --- a/test/tests.hs +++ b/test/tests.hs @@ -2,23 +2,42 @@ module Main where -import Data.Functor +import Prelude + +-- base import Data.Either (isRight, isLeft) import qualified Data.List.NonEmpty as NE + +-- Hackage import Control.Lens import Control.Monad.Except (catchError, throwError) import Control.Monad.Trans (liftIO) -import Network.Kafka -import Network.Kafka.Consumer -import Network.Kafka.Producer -import Network.Kafka.Protocol (ProduceResponse(..), KafkaError(..), CompressionCodec(..), CreateTopicsResponse(..)) import Test.Tasty import Test.Tasty.Hspec import Test.Tasty.QuickCheck import qualified Data.ByteString.Char8 as B -import Prelude +-- local +import Network.Kafka +import Network.Kafka.Consumer +import Network.Kafka.Producer +import Network.Kafka.Protocol (ProduceResponse(..), + KafkaError(..), + CompressionCodec(..), + CreateTopicsResponse(..), + DeleteTopicsResponse(..), + Offset(..), + OffsetCommitResponse(..), + OffsetFetchResponse(..), + ConsumerGroup(..), + Partition(..), + KafkaString(..), + Metadata(..), + TopicName(..), + ) + + main :: IO () main = testSpec "the specs" specs >>= defaultMain @@ -32,6 +51,13 @@ stateWaitTime .= 1000 byteMessages = fmap (TopicAndMessage topic . makeMessage . B.pack) + let cleanup :: TopicName -> IO () + cleanup topicName = do + run $ do + stateAddresses %= NE.cons ("localhost", 9092) + deleteTopic (deleteTopicsRequest topicName) + pure () + describe "can talk to local Kafka server" $ do prop "can produce messages" $ \ms -> do result <- run . produceMessages $ byteMessages ms @@ -129,13 +155,76 @@ use stateAddresses result `shouldBe` fmap NE.nub result - describe "create topics" $ + + let newTopicName = "milena-test-13-partitions" + describe "create topics" $ do it "create topics with multiple partitions" $ do - let t = "milena-test-13-partitions" result <- run $ do stateAddresses %= NE.cons ("localhost", 9092) - createTopic (createTopicsRequest t 13 1 [] []) - result `shouldBe` (Right $ TopicsResp [(t, NoError)]) + createTopic (createTopicsRequest newTopicName 13 1 [] []) + result `shouldBe` (Right $ TopicsResp [(newTopicName, NoError)]) + + it "create already existing topic" $ do + result <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + createTopic (createTopicsRequest newTopicName 13 1 [] []) + result `shouldBe` (Right $ TopicsResp [(newTopicName, TopicAlreadyExists)]) + + describe "delete topics" $ + it "delete topics" $ do + result <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + deleteTopic (deleteTopicsRequest newTopicName) + result `shouldBe` (Right $ DeleteTopicsResp [(newTopicName, NoError)]) + + + let commitOffsetTopicName = "commit-offset" + t = commitOffsetTopicName + + Test.Tasty.Hspec.afterAll_ (cleanup commitOffsetTopicName) $ do + describe "can commit messages" $ do + it "create a topic" $ do + topicCreation <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + createTopic (createTopicsRequest t 3 1 [] []) + topicCreation `shouldBe` (Right $ TopicsResp [(t, NoError)]) + + it "commit offset 5 to partition 0 for consumer group \"group1\"" $ do + commitOff <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + commitOffset (commitOffsetRequest (ConsumerGroup "group1") t 0 (Offset 5)) + commitOff `shouldBe` Right (OffsetCommitResp [(t,[(Partition 0,NoError)])]) + + it "commit offset 15 to partition 1 for consumer group \"group1\"" $ do + commitOff <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + commitOffset (commitOffsetRequest (ConsumerGroup "group1") t 1 (Offset 15)) + commitOff `shouldBe` Right (OffsetCommitResp [(t,[(Partition 1,NoError)])]) + + it "commit offset 10 to partition 2 for consumer group \"group2\"" $ do + commitOff <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + commitOffset (commitOffsetRequest (ConsumerGroup "group2") t 2 10) + commitOff `shouldSatisfy` isRight + + it "fetch offset from partition 0 for \"group1\"" $ do + fetchOff <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + fetchOffset (fetchOffsetRequest (ConsumerGroup "group1") t 0) + fetchOff `shouldBe` Right (OffsetFetchResp [(t,[(Partition 0, Offset 5,Metadata (KString {_kString = ""}),NoError)])]) + + it "fetch offset from partition 0 for \"group2\"" $ do + fetchOff <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + fetchOffset (fetchOffsetRequest (ConsumerGroup "group2") t 0) + fetchOff `shouldBe` Right (OffsetFetchResp [(t,[(Partition 0, Offset (-1),Metadata (KString {_kString = ""}),UnknownTopicOrPartition)])]) + + it "note that getLastOffset is unchanged" $ do + getLastOff <- run $ do + stateAddresses %= NE.cons ("localhost", 9092) + getLastOffset EarliestTime 0 t + getLastOff `shouldSatisfy` isRight + getLastOff `shouldBe` (Right $ Offset 0) prop :: Testable prop => String -> prop -> SpecWith () prop s = it s . property