Kafka.Consumer

Module to consume messages from Kafka topics.

Here's an example of code to consume messages from a topic:

import Control.Exception (bracket)
import Control.Monad (replicateM_)
import Kafka.Consumer

-- Global consumer properties
consumerProps :: ConsumerProperties
consumerProps = brokersList ["localhost:9092"]
             <> groupId (ConsumerGroupId "consumer_example_group")
             <> noAutoCommit
             <> logLevel KafkaLogInfo

-- Subscription to topics
consumerSub :: Subscription
consumerSub = topics [TopicName "kafka-client-example-topic"]
           <> offsetReset Earliest

-- Running an example
runConsumerExample :: IO ()
runConsumerExample = do
    res <- bracket mkConsumer clConsumer runHandler
    print res
    where
      mkConsumer = newConsumer consumerProps consumerSub
      clConsumer (Left err) = pure (Left err)
      clConsumer (Right kc) = (maybe (Right ()) Left) <$> closeConsumer kc
      runHandler (Left err) = pure (Left err)
      runHandler (Right kc) = processMessages kc

-- Example polling 10 times before stopping
processMessages :: KafkaConsumer -> IO (Either KafkaError ())
processMessages kafka = do
    replicateM_ 10 $ do
      msg <- pollMessage kafka (Timeout 1000)
      putStrLn $ "Message: " <> show msg
      err <- commitAllOffsets OffsetCommit kafka
      putStrLn $ "Offsets: " <> maybe "Committed." show err
    pure $ Right ()

Documentation

data KafkaConsumer

The main type for Kafka consumption, used e.g. to poll and commit messages.

Its constructor is intentionally not exposed, instead, one should use newConsumer to acquire such a value.

data KafkaCompressionCodec

Compression codec used by a topic

See Kafka documentation on compression codecs

Constructors

NoCompression

Gzip

Snappy

Lz4

Instances

Eq KafkaCompressionCodec

Defined in Kafka.Types

Methods

(==) :: KafkaCompressionCodec -> KafkaCompressionCodec -> Bool

(/=) :: KafkaCompressionCodec -> KafkaCompressionCodec -> Bool

Show KafkaCompressionCodec

Defined in Kafka.Types

Methods

showsPrec :: Int -> KafkaCompressionCodec -> ShowS

show :: KafkaCompressionCodec -> String

showList :: [KafkaCompressionCodec] -> ShowS

Generic KafkaCompressionCodec

Defined in Kafka.Types

Associated Types

type Rep KafkaCompressionCodec :: Type -> Type

Methods

from :: KafkaCompressionCodec -> Rep KafkaCompressionCodec x

to :: Rep KafkaCompressionCodec x -> KafkaCompressionCodec

type Rep KafkaCompressionCodec

Defined in Kafka.Types

type Rep KafkaCompressionCodec = D1 ('MetaData "KafkaCompressionCodec" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'False) ((C1 ('MetaCons "NoCompression" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "Gzip" 'PrefixI 'False) (U1 :: Type -> Type)) :+: (C1 ('MetaCons "Snappy" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "Lz4" 'PrefixI 'False) (U1 :: Type -> Type)))

data KafkaDebug

Available librdkafka debug contexts

Constructors

DebugGeneric

DebugBroker

DebugTopic

DebugMetadata

DebugQueue

DebugMsg

DebugProtocol

DebugCgrp

DebugSecurity

DebugFetch

DebugFeature

DebugAll

Instances

Eq KafkaDebug

Defined in Kafka.Types

Methods

(==) :: KafkaDebug -> KafkaDebug -> Bool

(/=) :: KafkaDebug -> KafkaDebug -> Bool

Show KafkaDebug

Defined in Kafka.Types

Methods

showsPrec :: Int -> KafkaDebug -> ShowS

show :: KafkaDebug -> String

showList :: [KafkaDebug] -> ShowS

Generic KafkaDebug

Defined in Kafka.Types

Associated Types

type Rep KafkaDebug :: Type -> Type

Methods

from :: KafkaDebug -> Rep KafkaDebug x

to :: Rep KafkaDebug x -> KafkaDebug

type Rep KafkaDebug

Defined in Kafka.Types

type Rep KafkaDebug = D1 ('MetaData "KafkaDebug" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'False) (((C1 ('MetaCons "DebugGeneric" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "DebugBroker" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "DebugTopic" 'PrefixI 'False) (U1 :: Type -> Type))) :+: (C1 ('MetaCons "DebugMetadata" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "DebugQueue" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "DebugMsg" 'PrefixI 'False) (U1 :: Type -> Type)))) :+: ((C1 ('MetaCons "DebugProtocol" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "DebugCgrp" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "DebugSecurity" 'PrefixI 'False) (U1 :: Type -> Type))) :+: (C1 ('MetaCons "DebugFetch" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "DebugFeature" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "DebugAll" 'PrefixI 'False) (U1 :: Type -> Type)))))

data KafkaError

All possible Kafka errors

Constructors

KafkaError Text

KafkaInvalidReturnValue

KafkaBadSpecification Text

KafkaResponseError RdKafkaRespErrT

KafkaInvalidConfigurationValue Text

KafkaUnknownConfigurationKey Text

KafkaBadConfiguration

Instances

Eq KafkaError

Defined in Kafka.Types

Methods

(==) :: KafkaError -> KafkaError -> Bool

(/=) :: KafkaError -> KafkaError -> Bool

Show KafkaError

Defined in Kafka.Types

Methods

showsPrec :: Int -> KafkaError -> ShowS

show :: KafkaError -> String

showList :: [KafkaError] -> ShowS

Generic KafkaError

Defined in Kafka.Types

Associated Types

type Rep KafkaError :: Type -> Type

Methods

from :: KafkaError -> Rep KafkaError x

to :: Rep KafkaError x -> KafkaError

Exception KafkaError

Defined in Kafka.Types

Methods

toException :: KafkaError -> SomeException

fromException :: SomeException -> Maybe KafkaError

displayException :: KafkaError -> String

type Rep KafkaError

Defined in Kafka.Types

type Rep KafkaError = D1 ('MetaData "KafkaError" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'False) ((C1 ('MetaCons "KafkaError" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)) :+: (C1 ('MetaCons "KafkaInvalidReturnValue" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "KafkaBadSpecification" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)))) :+: ((C1 ('MetaCons "KafkaResponseError" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 RdKafkaRespErrT)) :+: C1 ('MetaCons "KafkaInvalidConfigurationValue" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text))) :+: (C1 ('MetaCons "KafkaUnknownConfigurationKey" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)) :+: C1 ('MetaCons "KafkaBadConfiguration" 'PrefixI 'False) (U1 :: Type -> Type))))

data KafkaLogLevel

Log levels for librdkafka.

Constructors

KafkaLogEmerg

KafkaLogAlert

KafkaLogCrit

KafkaLogErr

KafkaLogWarning

KafkaLogNotice

KafkaLogInfo

KafkaLogDebug

Instances

Enum KafkaLogLevel

Defined in Kafka.Types

Methods

succ :: KafkaLogLevel -> KafkaLogLevel

pred :: KafkaLogLevel -> KafkaLogLevel

toEnum :: Int -> KafkaLogLevel

fromEnum :: KafkaLogLevel -> Int

enumFrom :: KafkaLogLevel -> [KafkaLogLevel]

enumFromThen :: KafkaLogLevel -> KafkaLogLevel -> [KafkaLogLevel]

enumFromTo :: KafkaLogLevel -> KafkaLogLevel -> [KafkaLogLevel]

enumFromThenTo :: KafkaLogLevel -> KafkaLogLevel -> KafkaLogLevel -> [KafkaLogLevel]

Eq KafkaLogLevel

Defined in Kafka.Types

Methods

(==) :: KafkaLogLevel -> KafkaLogLevel -> Bool

(/=) :: KafkaLogLevel -> KafkaLogLevel -> Bool

Show KafkaLogLevel

Defined in Kafka.Types

Methods

showsPrec :: Int -> KafkaLogLevel -> ShowS

show :: KafkaLogLevel -> String

showList :: [KafkaLogLevel] -> ShowS

newtype Timeout

Timeout in milliseconds

Constructors

Timeout

Fields

• unTimeout :: Int

Instances

Eq Timeout

Defined in Kafka.Types

Methods

(==) :: Timeout -> Timeout -> Bool

(/=) :: Timeout -> Timeout -> Bool

Read Timeout

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS Timeout

readList :: ReadS [Timeout]

readPrec :: ReadPrec Timeout

readListPrec :: ReadPrec [Timeout]

Show Timeout

Defined in Kafka.Types

Methods

showsPrec :: Int -> Timeout -> ShowS

show :: Timeout -> String

showList :: [Timeout] -> ShowS

Generic Timeout

Defined in Kafka.Types

Associated Types

type Rep Timeout :: Type -> Type

Methods

from :: Timeout -> Rep Timeout x

to :: Rep Timeout x -> Timeout

type Rep Timeout

Defined in Kafka.Types

type Rep Timeout = D1 ('MetaData "Timeout" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "Timeout" 'PrefixI 'True) (S1 ('MetaSel ('Just "unTimeout") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int)))

newtype BrokerAddress

Kafka broker address string (e.g. broker1:9092)

Constructors

BrokerAddress

Fields

• unBrokerAddress :: Text

Instances

Eq BrokerAddress

Defined in Kafka.Types

Methods

(==) :: BrokerAddress -> BrokerAddress -> Bool

(/=) :: BrokerAddress -> BrokerAddress -> Bool

Show BrokerAddress

Defined in Kafka.Types

Methods

showsPrec :: Int -> BrokerAddress -> ShowS

show :: BrokerAddress -> String

showList :: [BrokerAddress] -> ShowS

IsString BrokerAddress

Defined in Kafka.Types

Methods

fromString :: String -> BrokerAddress

Generic BrokerAddress

Defined in Kafka.Types

Associated Types

type Rep BrokerAddress :: Type -> Type

Methods

from :: BrokerAddress -> Rep BrokerAddress x

to :: Rep BrokerAddress x -> BrokerAddress

type Rep BrokerAddress

Defined in Kafka.Types

type Rep BrokerAddress = D1 ('MetaData "BrokerAddress" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "BrokerAddress" 'PrefixI 'True) (S1 ('MetaSel ('Just "unBrokerAddress") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)))

newtype TopicName

Topic name to consume/produce messages

Wildcard (regex) topics are supported by the librdkafka assignor: any topic name in the topics list that is prefixed with ^ will be regex-matched to the full list of topics in the cluster and matching topics will be added to the subscription list.

Constructors

TopicName

Fields

• unTopicName :: Text

  a simple topic name or a regex if started with ^

Instances

Eq TopicName

Defined in Kafka.Types

Methods

(==) :: TopicName -> TopicName -> Bool

(/=) :: TopicName -> TopicName -> Bool

Ord TopicName

Defined in Kafka.Types

Methods

compare :: TopicName -> TopicName -> Ordering

(<) :: TopicName -> TopicName -> Bool

(<=) :: TopicName -> TopicName -> Bool

(>) :: TopicName -> TopicName -> Bool

(>=) :: TopicName -> TopicName -> Bool

max :: TopicName -> TopicName -> TopicName

min :: TopicName -> TopicName -> TopicName

Read TopicName

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS TopicName

readList :: ReadS [TopicName]

readPrec :: ReadPrec TopicName

readListPrec :: ReadPrec [TopicName]

Show TopicName

Defined in Kafka.Types

Methods

showsPrec :: Int -> TopicName -> ShowS

show :: TopicName -> String

showList :: [TopicName] -> ShowS

IsString TopicName

Defined in Kafka.Types

Methods

fromString :: String -> TopicName

Generic TopicName

Defined in Kafka.Types

Associated Types

type Rep TopicName :: Type -> Type

Methods

from :: TopicName -> Rep TopicName x

to :: Rep TopicName x -> TopicName

type Rep TopicName

Defined in Kafka.Types

type Rep TopicName = D1 ('MetaData "TopicName" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "TopicName" 'PrefixI 'True) (S1 ('MetaSel ('Just "unTopicName") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)))

data TopicType

Whether the topic is created by a user or by the system

Constructors

User

Normal topics that are created by user.

System

Topics starting with a double underscore "__" (__consumer_offsets, __confluent.support.metrics, etc.) are considered "system" topics

Instances

Eq TopicType

Defined in Kafka.Types

Methods

(==) :: TopicType -> TopicType -> Bool

(/=) :: TopicType -> TopicType -> Bool

Ord TopicType

Defined in Kafka.Types

Methods

compare :: TopicType -> TopicType -> Ordering

(<) :: TopicType -> TopicType -> Bool

(<=) :: TopicType -> TopicType -> Bool

(>) :: TopicType -> TopicType -> Bool

(>=) :: TopicType -> TopicType -> Bool

max :: TopicType -> TopicType -> TopicType

min :: TopicType -> TopicType -> TopicType

Read TopicType

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS TopicType

readList :: ReadS [TopicType]

readPrec :: ReadPrec TopicType

readListPrec :: ReadPrec [TopicType]

Show TopicType

Defined in Kafka.Types

Methods

showsPrec :: Int -> TopicType -> ShowS

show :: TopicType -> String

showList :: [TopicType] -> ShowS

Generic TopicType

Defined in Kafka.Types

Associated Types

type Rep TopicType :: Type -> Type

Methods

from :: TopicType -> Rep TopicType x

to :: Rep TopicType x -> TopicType

type Rep TopicType

Defined in Kafka.Types

type Rep TopicType = D1 ('MetaData "TopicType" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "User" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "System" 'PrefixI 'False) (U1 :: Type -> Type))

newtype BatchSize

Batch size used for polling

Constructors

BatchSize

Fields

• unBatchSize :: Int

Instances

Eq BatchSize

Defined in Kafka.Types

Methods

(==) :: BatchSize -> BatchSize -> Bool

(/=) :: BatchSize -> BatchSize -> Bool

Num BatchSize

Defined in Kafka.Types

Methods

(+) :: BatchSize -> BatchSize -> BatchSize

(-) :: BatchSize -> BatchSize -> BatchSize

(*) :: BatchSize -> BatchSize -> BatchSize

negate :: BatchSize -> BatchSize

abs :: BatchSize -> BatchSize

signum :: BatchSize -> BatchSize

fromInteger :: Integer -> BatchSize

Ord BatchSize

Defined in Kafka.Types

Methods

compare :: BatchSize -> BatchSize -> Ordering

(<) :: BatchSize -> BatchSize -> Bool

(<=) :: BatchSize -> BatchSize -> Bool

(>) :: BatchSize -> BatchSize -> Bool

(>=) :: BatchSize -> BatchSize -> Bool

max :: BatchSize -> BatchSize -> BatchSize

min :: BatchSize -> BatchSize -> BatchSize

Read BatchSize

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS BatchSize

readList :: ReadS [BatchSize]

readPrec :: ReadPrec BatchSize

readListPrec :: ReadPrec [BatchSize]

Show BatchSize

Defined in Kafka.Types

Methods

showsPrec :: Int -> BatchSize -> ShowS

show :: BatchSize -> String

showList :: [BatchSize] -> ShowS

Generic BatchSize

Defined in Kafka.Types

Associated Types

type Rep BatchSize :: Type -> Type

Methods

from :: BatchSize -> Rep BatchSize x

to :: Rep BatchSize x -> BatchSize

type Rep BatchSize

Defined in Kafka.Types

type Rep BatchSize = D1 ('MetaData "BatchSize" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "BatchSize" 'PrefixI 'True) (S1 ('MetaSel ('Just "unBatchSize") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int)))

newtype ClientId

Client ID used by Kafka to better track requests

See Kafka documentation on client ID

Constructors

ClientId

Fields

• unClientId :: Text

Instances

Eq ClientId

Defined in Kafka.Types

Methods

(==) :: ClientId -> ClientId -> Bool

(/=) :: ClientId -> ClientId -> Bool

Ord ClientId

Defined in Kafka.Types

Methods

compare :: ClientId -> ClientId -> Ordering

(<) :: ClientId -> ClientId -> Bool

(<=) :: ClientId -> ClientId -> Bool

(>) :: ClientId -> ClientId -> Bool

(>=) :: ClientId -> ClientId -> Bool

max :: ClientId -> ClientId -> ClientId

min :: ClientId -> ClientId -> ClientId

Show ClientId

Defined in Kafka.Types

Methods

showsPrec :: Int -> ClientId -> ShowS

show :: ClientId -> String

showList :: [ClientId] -> ShowS

IsString ClientId

Defined in Kafka.Types

Methods

fromString :: String -> ClientId

Generic ClientId

Defined in Kafka.Types

Associated Types

type Rep ClientId :: Type -> Type

Methods

from :: ClientId -> Rep ClientId x

to :: Rep ClientId x -> ClientId

type Rep ClientId

Defined in Kafka.Types

type Rep ClientId = D1 ('MetaData "ClientId" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "ClientId" 'PrefixI 'True) (S1 ('MetaSel ('Just "unClientId") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)))

newtype Millis

A number of milliseconds, used to represent durations and timestamps

Constructors

Millis

Fields

• unMillis :: Int64

Instances

Eq Millis

Defined in Kafka.Types

Methods

(==) :: Millis -> Millis -> Bool

(/=) :: Millis -> Millis -> Bool

Num Millis

Defined in Kafka.Types

Methods

(+) :: Millis -> Millis -> Millis

(-) :: Millis -> Millis -> Millis

(*) :: Millis -> Millis -> Millis

negate :: Millis -> Millis

abs :: Millis -> Millis

signum :: Millis -> Millis

fromInteger :: Integer -> Millis

Ord Millis

Defined in Kafka.Types

Methods

compare :: Millis -> Millis -> Ordering

(<) :: Millis -> Millis -> Bool

(<=) :: Millis -> Millis -> Bool

(>) :: Millis -> Millis -> Bool

(>=) :: Millis -> Millis -> Bool

max :: Millis -> Millis -> Millis

min :: Millis -> Millis -> Millis

Read Millis

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS Millis

readList :: ReadS [Millis]

readPrec :: ReadPrec Millis

readListPrec :: ReadPrec [Millis]

Show Millis

Defined in Kafka.Types

Methods

showsPrec :: Int -> Millis -> ShowS

show :: Millis -> String

showList :: [Millis] -> ShowS

Generic Millis

Defined in Kafka.Types

Associated Types

type Rep Millis :: Type -> Type

Methods

from :: Millis -> Rep Millis x

to :: Rep Millis x -> Millis

type Rep Millis

Defined in Kafka.Types

type Rep Millis = D1 ('MetaData "Millis" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "Millis" 'PrefixI 'True) (S1 ('MetaSel ('Just "unMillis") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64)))

newtype PartitionId

Topic partition ID

Constructors

PartitionId

Fields

• unPartitionId :: Int

Instances

Enum PartitionId

Defined in Kafka.Types

Methods

succ :: PartitionId -> PartitionId

pred :: PartitionId -> PartitionId

toEnum :: Int -> PartitionId

fromEnum :: PartitionId -> Int

enumFrom :: PartitionId -> [PartitionId]

enumFromThen :: PartitionId -> PartitionId -> [PartitionId]

enumFromTo :: PartitionId -> PartitionId -> [PartitionId]

enumFromThenTo :: PartitionId -> PartitionId -> PartitionId -> [PartitionId]

Eq PartitionId

Defined in Kafka.Types

Methods

(==) :: PartitionId -> PartitionId -> Bool

(/=) :: PartitionId -> PartitionId -> Bool

Ord PartitionId

Defined in Kafka.Types

Methods

compare :: PartitionId -> PartitionId -> Ordering

(<) :: PartitionId -> PartitionId -> Bool

(<=) :: PartitionId -> PartitionId -> Bool

(>) :: PartitionId -> PartitionId -> Bool

(>=) :: PartitionId -> PartitionId -> Bool

max :: PartitionId -> PartitionId -> PartitionId

min :: PartitionId -> PartitionId -> PartitionId

Read PartitionId

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS PartitionId

readList :: ReadS [PartitionId]

readPrec :: ReadPrec PartitionId

readListPrec :: ReadPrec [PartitionId]

Show PartitionId

Defined in Kafka.Types

Methods

showsPrec :: Int -> PartitionId -> ShowS

show :: PartitionId -> String

showList :: [PartitionId] -> ShowS

Generic PartitionId

Defined in Kafka.Types

Associated Types

type Rep PartitionId :: Type -> Type

Methods

from :: PartitionId -> Rep PartitionId x

to :: Rep PartitionId x -> PartitionId

type Rep PartitionId

Defined in Kafka.Types

type Rep PartitionId = D1 ('MetaData "PartitionId" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "PartitionId" 'PrefixI 'True) (S1 ('MetaSel ('Just "unPartitionId") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int)))

newtype BrokerId

Kafka broker ID

Constructors

BrokerId

Fields

• unBrokerId :: Int

Instances

Eq BrokerId

Defined in Kafka.Types

Methods

(==) :: BrokerId -> BrokerId -> Bool

(/=) :: BrokerId -> BrokerId -> Bool

Ord BrokerId

Defined in Kafka.Types

Methods

compare :: BrokerId -> BrokerId -> Ordering

(<) :: BrokerId -> BrokerId -> Bool

(<=) :: BrokerId -> BrokerId -> Bool

(>) :: BrokerId -> BrokerId -> Bool

(>=) :: BrokerId -> BrokerId -> Bool

max :: BrokerId -> BrokerId -> BrokerId

min :: BrokerId -> BrokerId -> BrokerId

Read BrokerId

Defined in Kafka.Types

Methods

readsPrec :: Int -> ReadS BrokerId

readList :: ReadS [BrokerId]

readPrec :: ReadPrec BrokerId

readListPrec :: ReadPrec [BrokerId]

Show BrokerId

Defined in Kafka.Types

Methods

showsPrec :: Int -> BrokerId -> ShowS

show :: BrokerId -> String

showList :: [BrokerId] -> ShowS

Generic BrokerId

Defined in Kafka.Types

Associated Types

type Rep BrokerId :: Type -> Type

Methods

from :: BrokerId -> Rep BrokerId x

to :: Rep BrokerId x -> BrokerId

type Rep BrokerId

Defined in Kafka.Types

type Rep BrokerId = D1 ('MetaData "BrokerId" "Kafka.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "BrokerId" 'PrefixI 'True) (S1 ('MetaSel ('Just "unBrokerId") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int)))

topicType :: TopicName -> TopicType

Deduce the type of a topic from its name, by checking if it starts with a double underscore "__"

kafkaDebugToText :: KafkaDebug -> Text

Convert a KafkaDebug into its librdkafka string equivalent.

This is used internally by the library but may be useful to some developers.

kafkaCompressionCodecToText :: KafkaCompressionCodec -> Text

Convert a KafkaCompressionCodec into its librdkafka string equivalent.

This is used internally by the library but may be useful to some developers.

data Callback

Callbacks allow retrieving various information like error occurences, statistics and log messages. See setCallback (Consumer) and setCallback (Producer) for more details.

data ConsumerRecord k v

Represents a received message from Kafka (i.e. used in a consumer)

Constructors

ConsumerRecord

Fields

• crTopic :: !TopicName

  Kafka topic this message was received from

• crPartition :: !PartitionId

  Kafka partition this message was received from

• crOffset :: !Offset

  Offset within the crPartition Kafka partition

• crTimestamp :: !Timestamp

  Message timestamp

• crKey :: !k

  Message key

• crValue :: !v

  Message value

Instances

Bitraversable ConsumerRecord

Defined in Kafka.Consumer.Types

Methods

bitraverse :: Applicative f => (a -> f c) -> (b -> f d) -> ConsumerRecord a b -> f (ConsumerRecord c d)

Bifoldable ConsumerRecord

Defined in Kafka.Consumer.Types

Methods

bifold :: Monoid m => ConsumerRecord m m -> m

bifoldMap :: Monoid m => (a -> m) -> (b -> m) -> ConsumerRecord a b -> m

bifoldr :: (a -> c -> c) -> (b -> c -> c) -> c -> ConsumerRecord a b -> c

bifoldl :: (c -> a -> c) -> (c -> b -> c) -> c -> ConsumerRecord a b -> c

Bifunctor ConsumerRecord

Defined in Kafka.Consumer.Types

Methods

bimap :: (a -> b) -> (c -> d) -> ConsumerRecord a c -> ConsumerRecord b d

first :: (a -> b) -> ConsumerRecord a c -> ConsumerRecord b c

second :: (b -> c) -> ConsumerRecord a b -> ConsumerRecord a c

Functor (ConsumerRecord k)

Defined in Kafka.Consumer.Types

Methods

fmap :: (a -> b) -> ConsumerRecord k a -> ConsumerRecord k b

(<$) :: a -> ConsumerRecord k b -> ConsumerRecord k a

Foldable (ConsumerRecord k)

Defined in Kafka.Consumer.Types

Methods

fold :: Monoid m => ConsumerRecord k m -> m

foldMap :: Monoid m => (a -> m) -> ConsumerRecord k a -> m

foldMap' :: Monoid m => (a -> m) -> ConsumerRecord k a -> m

foldr :: (a -> b -> b) -> b -> ConsumerRecord k a -> b

foldr' :: (a -> b -> b) -> b -> ConsumerRecord k a -> b

foldl :: (b -> a -> b) -> b -> ConsumerRecord k a -> b

foldl' :: (b -> a -> b) -> b -> ConsumerRecord k a -> b

foldr1 :: (a -> a -> a) -> ConsumerRecord k a -> a

foldl1 :: (a -> a -> a) -> ConsumerRecord k a -> a

toList :: ConsumerRecord k a -> [a]

null :: ConsumerRecord k a -> Bool

length :: ConsumerRecord k a -> Int

elem :: Eq a => a -> ConsumerRecord k a -> Bool

maximum :: Ord a => ConsumerRecord k a -> a

minimum :: Ord a => ConsumerRecord k a -> a

sum :: Num a => ConsumerRecord k a -> a

product :: Num a => ConsumerRecord k a -> a

Traversable (ConsumerRecord k)

Defined in Kafka.Consumer.Types

Methods

traverse :: Applicative f => (a -> f b) -> ConsumerRecord k a -> f (ConsumerRecord k b)

sequenceA :: Applicative f => ConsumerRecord k (f a) -> f (ConsumerRecord k a)

mapM :: Monad m => (a -> m b) -> ConsumerRecord k a -> m (ConsumerRecord k b)

sequence :: Monad m => ConsumerRecord k (m a) -> m (ConsumerRecord k a)

(Eq k, Eq v) => Eq (ConsumerRecord k v)

Defined in Kafka.Consumer.Types

Methods

(==) :: ConsumerRecord k v -> ConsumerRecord k v -> Bool

(/=) :: ConsumerRecord k v -> ConsumerRecord k v -> Bool

(Read k, Read v) => Read (ConsumerRecord k v)

Defined in Kafka.Consumer.Types

Methods

readsPrec :: Int -> ReadS (ConsumerRecord k v)

readList :: ReadS [ConsumerRecord k v]

readPrec :: ReadPrec (ConsumerRecord k v)

readListPrec :: ReadPrec [ConsumerRecord k v]

(Show k, Show v) => Show (ConsumerRecord k v)

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> ConsumerRecord k v -> ShowS

show :: ConsumerRecord k v -> String

showList :: [ConsumerRecord k v] -> ShowS

Generic (ConsumerRecord k v)

Defined in Kafka.Consumer.Types

Associated Types

type Rep (ConsumerRecord k v) :: Type -> Type

Methods

from :: ConsumerRecord k v -> Rep (ConsumerRecord k v) x

to :: Rep (ConsumerRecord k v) x -> ConsumerRecord k v

type Rep (ConsumerRecord k v)

Defined in Kafka.Consumer.Types

type Rep (ConsumerRecord k v) = D1 ('MetaData "ConsumerRecord" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "ConsumerRecord" 'PrefixI 'True) ((S1 ('MetaSel ('Just "crTopic") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 TopicName) :*: (S1 ('MetaSel ('Just "crPartition") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 PartitionId) :*: S1 ('MetaSel ('Just "crOffset") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Offset))) :*: (S1 ('MetaSel ('Just "crTimestamp") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Timestamp) :*: (S1 ('MetaSel ('Just "crKey") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 k) :*: S1 ('MetaSel ('Just "crValue") 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 v)))))

data TopicPartition

Kafka topic partition structure

Constructors

TopicPartition

Fields

• tpTopicName :: TopicName
• tpPartition :: PartitionId
• tpOffset :: PartitionOffset

Instances

Eq TopicPartition

Defined in Kafka.Consumer.Types

Methods

(==) :: TopicPartition -> TopicPartition -> Bool

(/=) :: TopicPartition -> TopicPartition -> Bool

Show TopicPartition

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> TopicPartition -> ShowS

show :: TopicPartition -> String

showList :: [TopicPartition] -> ShowS

Generic TopicPartition

Defined in Kafka.Consumer.Types

Associated Types

type Rep TopicPartition :: Type -> Type

Methods

from :: TopicPartition -> Rep TopicPartition x

to :: Rep TopicPartition x -> TopicPartition

type Rep TopicPartition

Defined in Kafka.Consumer.Types

type Rep TopicPartition = D1 ('MetaData "TopicPartition" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "TopicPartition" 'PrefixI 'True) (S1 ('MetaSel ('Just "tpTopicName") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 TopicName) :*: (S1 ('MetaSel ('Just "tpPartition") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 PartitionId) :*: S1 ('MetaSel ('Just "tpOffset") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 PartitionOffset))))

data OffsetStoreMethod

Indicates the method of storing the offsets

Constructors

OffsetStoreBroker

Offsets are stored in Kafka broker (preferred)

OffsetStoreFile FilePath OffsetStoreSync

Offsets are stored in a file (and synced to disk according to the sync policy)

Instances

Eq OffsetStoreMethod

Defined in Kafka.Consumer.Types

Methods

(==) :: OffsetStoreMethod -> OffsetStoreMethod -> Bool

(/=) :: OffsetStoreMethod -> OffsetStoreMethod -> Bool

Show OffsetStoreMethod

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> OffsetStoreMethod -> ShowS

show :: OffsetStoreMethod -> String

showList :: [OffsetStoreMethod] -> ShowS

Generic OffsetStoreMethod

Defined in Kafka.Consumer.Types

Associated Types

type Rep OffsetStoreMethod :: Type -> Type

Methods

from :: OffsetStoreMethod -> Rep OffsetStoreMethod x

to :: Rep OffsetStoreMethod x -> OffsetStoreMethod

type Rep OffsetStoreMethod

Defined in Kafka.Consumer.Types

type Rep OffsetStoreMethod = D1 ('MetaData "OffsetStoreMethod" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "OffsetStoreBroker" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "OffsetStoreFile" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 FilePath) :*: S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 OffsetStoreSync)))

data OffsetStoreSync

Indicates how offsets are to be synced to disk

Constructors

OffsetSyncDisable

Do not sync offsets (in Kafka: -1)

OffsetSyncImmediate

Sync immediately after each offset commit (in Kafka: 0)

OffsetSyncInterval Int

Sync after specified interval in millis

Instances

Eq OffsetStoreSync

Defined in Kafka.Consumer.Types

Methods

(==) :: OffsetStoreSync -> OffsetStoreSync -> Bool

(/=) :: OffsetStoreSync -> OffsetStoreSync -> Bool

Show OffsetStoreSync

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> OffsetStoreSync -> ShowS

show :: OffsetStoreSync -> String

showList :: [OffsetStoreSync] -> ShowS

Generic OffsetStoreSync

Defined in Kafka.Consumer.Types

Associated Types

type Rep OffsetStoreSync :: Type -> Type

Methods

from :: OffsetStoreSync -> Rep OffsetStoreSync x

to :: Rep OffsetStoreSync x -> OffsetStoreSync

type Rep OffsetStoreSync

Defined in Kafka.Consumer.Types

type Rep OffsetStoreSync = D1 ('MetaData "OffsetStoreSync" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "OffsetSyncDisable" 'PrefixI 'False) (U1 :: Type -> Type) :+: (C1 ('MetaCons "OffsetSyncImmediate" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "OffsetSyncInterval" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int))))

data OffsetCommit

Offsets commit mode

Constructors

OffsetCommit

Forces consumer to block until the broker offsets commit is done

OffsetCommitAsync

Offsets will be committed in a non-blocking way

Instances

Eq OffsetCommit

Defined in Kafka.Consumer.Types

Methods

(==) :: OffsetCommit -> OffsetCommit -> Bool

(/=) :: OffsetCommit -> OffsetCommit -> Bool

Show OffsetCommit

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> OffsetCommit -> ShowS

show :: OffsetCommit -> String

showList :: [OffsetCommit] -> ShowS

Generic OffsetCommit

Defined in Kafka.Consumer.Types

Associated Types

type Rep OffsetCommit :: Type -> Type

Methods

from :: OffsetCommit -> Rep OffsetCommit x

to :: Rep OffsetCommit x -> OffsetCommit

type Rep OffsetCommit

Defined in Kafka.Consumer.Types

type Rep OffsetCommit = D1 ('MetaData "OffsetCommit" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "OffsetCommit" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "OffsetCommitAsync" 'PrefixI 'False) (U1 :: Type -> Type))

data Timestamp

Consumer record timestamp

Constructors

CreateTime !Millis

LogAppendTime !Millis

NoTimestamp

Instances

Eq Timestamp

Defined in Kafka.Consumer.Types

Methods

(==) :: Timestamp -> Timestamp -> Bool

(/=) :: Timestamp -> Timestamp -> Bool

Read Timestamp

Defined in Kafka.Consumer.Types

Methods

readsPrec :: Int -> ReadS Timestamp

readList :: ReadS [Timestamp]

readPrec :: ReadPrec Timestamp

readListPrec :: ReadPrec [Timestamp]

Show Timestamp

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> Timestamp -> ShowS

show :: Timestamp -> String

showList :: [Timestamp] -> ShowS

Generic Timestamp

Defined in Kafka.Consumer.Types

Associated Types

type Rep Timestamp :: Type -> Type

Methods

from :: Timestamp -> Rep Timestamp x

to :: Rep Timestamp x -> Timestamp

type Rep Timestamp

Defined in Kafka.Consumer.Types

type Rep Timestamp = D1 ('MetaData "Timestamp" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "CreateTime" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Millis)) :+: (C1 ('MetaCons "LogAppendTime" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'SourceStrict 'DecidedStrict) (Rec0 Millis)) :+: C1 ('MetaCons "NoTimestamp" 'PrefixI 'False) (U1 :: Type -> Type)))

data SubscribedPartitions

Partitions subscribed by a consumer

Constructors

SubscribedPartitions [PartitionId]

Subscribe only to those partitions

SubscribedPartitionsAll

Subscribe to all partitions

Instances

Eq SubscribedPartitions

Defined in Kafka.Consumer.Types

Methods

(==) :: SubscribedPartitions -> SubscribedPartitions -> Bool

(/=) :: SubscribedPartitions -> SubscribedPartitions -> Bool

Show SubscribedPartitions

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> SubscribedPartitions -> ShowS

show :: SubscribedPartitions -> String

showList :: [SubscribedPartitions] -> ShowS

Generic SubscribedPartitions

Defined in Kafka.Consumer.Types

Associated Types

type Rep SubscribedPartitions :: Type -> Type

Methods

from :: SubscribedPartitions -> Rep SubscribedPartitions x

to :: Rep SubscribedPartitions x -> SubscribedPartitions

type Rep SubscribedPartitions

Defined in Kafka.Consumer.Types

type Rep SubscribedPartitions = D1 ('MetaData "SubscribedPartitions" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "SubscribedPartitions" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 [PartitionId])) :+: C1 ('MetaCons "SubscribedPartitionsAll" 'PrefixI 'False) (U1 :: Type -> Type))

data PartitionOffset

The partition offset

Constructors

PartitionOffsetBeginning

PartitionOffsetEnd

PartitionOffset Int64

PartitionOffsetStored

PartitionOffsetInvalid

Instances

Eq PartitionOffset

Defined in Kafka.Consumer.Types

Methods

(==) :: PartitionOffset -> PartitionOffset -> Bool

(/=) :: PartitionOffset -> PartitionOffset -> Bool

Show PartitionOffset

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> PartitionOffset -> ShowS

show :: PartitionOffset -> String

showList :: [PartitionOffset] -> ShowS

Generic PartitionOffset

Defined in Kafka.Consumer.Types

Associated Types

type Rep PartitionOffset :: Type -> Type

Methods

from :: PartitionOffset -> Rep PartitionOffset x

to :: Rep PartitionOffset x -> PartitionOffset

type Rep PartitionOffset

Defined in Kafka.Consumer.Types

type Rep PartitionOffset = D1 ('MetaData "PartitionOffset" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) ((C1 ('MetaCons "PartitionOffsetBeginning" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "PartitionOffsetEnd" 'PrefixI 'False) (U1 :: Type -> Type)) :+: (C1 ('MetaCons "PartitionOffset" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64)) :+: (C1 ('MetaCons "PartitionOffsetStored" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "PartitionOffsetInvalid" 'PrefixI 'False) (U1 :: Type -> Type))))

data RebalanceEvent

A set of events which happen during the rebalancing process

Constructors

RebalanceBeforeAssign [(TopicName, PartitionId)]

Happens before Kafka Client confirms new assignment

RebalanceAssign [(TopicName, PartitionId)]

Happens after the new assignment is confirmed

RebalanceBeforeRevoke [(TopicName, PartitionId)]

Happens before Kafka Client confirms partitions rejection

RebalanceRevoke [(TopicName, PartitionId)]

Happens after the rejection is confirmed

Instances

Eq RebalanceEvent

Defined in Kafka.Consumer.Types

Methods

(==) :: RebalanceEvent -> RebalanceEvent -> Bool

(/=) :: RebalanceEvent -> RebalanceEvent -> Bool

Show RebalanceEvent

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> RebalanceEvent -> ShowS

show :: RebalanceEvent -> String

showList :: [RebalanceEvent] -> ShowS

Generic RebalanceEvent

Defined in Kafka.Consumer.Types

Associated Types

type Rep RebalanceEvent :: Type -> Type

Methods

from :: RebalanceEvent -> Rep RebalanceEvent x

to :: Rep RebalanceEvent x -> RebalanceEvent

type Rep RebalanceEvent

Defined in Kafka.Consumer.Types

type Rep RebalanceEvent = D1 ('MetaData "RebalanceEvent" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) ((C1 ('MetaCons "RebalanceBeforeAssign" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 [(TopicName, PartitionId)])) :+: C1 ('MetaCons "RebalanceAssign" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 [(TopicName, PartitionId)]))) :+: (C1 ('MetaCons "RebalanceBeforeRevoke" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 [(TopicName, PartitionId)])) :+: C1 ('MetaCons "RebalanceRevoke" 'PrefixI 'False) (S1 ('MetaSel ('Nothing :: Maybe Symbol) 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 [(TopicName, PartitionId)]))))

data OffsetReset

Where to reset the offset when there is no initial offset in Kafka

See Kafka documentation on offset reset

Constructors

Earliest

Latest

Instances

Eq OffsetReset

Defined in Kafka.Consumer.Types

Methods

(==) :: OffsetReset -> OffsetReset -> Bool

(/=) :: OffsetReset -> OffsetReset -> Bool

Show OffsetReset

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> OffsetReset -> ShowS

show :: OffsetReset -> String

showList :: [OffsetReset] -> ShowS

Generic OffsetReset

Defined in Kafka.Consumer.Types

Associated Types

type Rep OffsetReset :: Type -> Type

Methods

from :: OffsetReset -> Rep OffsetReset x

to :: Rep OffsetReset x -> OffsetReset

type Rep OffsetReset

Defined in Kafka.Consumer.Types

type Rep OffsetReset = D1 ('MetaData "OffsetReset" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'False) (C1 ('MetaCons "Earliest" 'PrefixI 'False) (U1 :: Type -> Type) :+: C1 ('MetaCons "Latest" 'PrefixI 'False) (U1 :: Type -> Type))

newtype Offset

A message offset in a partition

Constructors

Offset

Fields

• unOffset :: Int64

Instances

Eq Offset

Defined in Kafka.Consumer.Types

Methods

(==) :: Offset -> Offset -> Bool

(/=) :: Offset -> Offset -> Bool

Ord Offset

Defined in Kafka.Consumer.Types

Methods

compare :: Offset -> Offset -> Ordering

(<) :: Offset -> Offset -> Bool

(<=) :: Offset -> Offset -> Bool

(>) :: Offset -> Offset -> Bool

(>=) :: Offset -> Offset -> Bool

max :: Offset -> Offset -> Offset

min :: Offset -> Offset -> Offset

Read Offset

Defined in Kafka.Consumer.Types

Methods

readsPrec :: Int -> ReadS Offset

readList :: ReadS [Offset]

readPrec :: ReadPrec Offset

readListPrec :: ReadPrec [Offset]

Show Offset

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> Offset -> ShowS

show :: Offset -> String

showList :: [Offset] -> ShowS

Generic Offset

Defined in Kafka.Consumer.Types

Associated Types

type Rep Offset :: Type -> Type

Methods

from :: Offset -> Rep Offset x

to :: Rep Offset x -> Offset

type Rep Offset

Defined in Kafka.Consumer.Types

type Rep Offset = D1 ('MetaData "Offset" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "Offset" 'PrefixI 'True) (S1 ('MetaSel ('Just "unOffset") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Int64)))

newtype ConsumerGroupId

Consumer group ID. Different consumers with the same consumer group ID will get assigned different partitions of each subscribed topic.

See Kafka documentation on consumer group

Constructors

ConsumerGroupId

Fields

• unConsumerGroupId :: Text

Instances

Eq ConsumerGroupId

Defined in Kafka.Consumer.Types

Methods

(==) :: ConsumerGroupId -> ConsumerGroupId -> Bool

(/=) :: ConsumerGroupId -> ConsumerGroupId -> Bool

Ord ConsumerGroupId

Defined in Kafka.Consumer.Types

Methods

compare :: ConsumerGroupId -> ConsumerGroupId -> Ordering

(<) :: ConsumerGroupId -> ConsumerGroupId -> Bool

(<=) :: ConsumerGroupId -> ConsumerGroupId -> Bool

(>) :: ConsumerGroupId -> ConsumerGroupId -> Bool

(>=) :: ConsumerGroupId -> ConsumerGroupId -> Bool

max :: ConsumerGroupId -> ConsumerGroupId -> ConsumerGroupId

min :: ConsumerGroupId -> ConsumerGroupId -> ConsumerGroupId

Show ConsumerGroupId

Defined in Kafka.Consumer.Types

Methods

showsPrec :: Int -> ConsumerGroupId -> ShowS

show :: ConsumerGroupId -> String

showList :: [ConsumerGroupId] -> ShowS

IsString ConsumerGroupId

Defined in Kafka.Consumer.Types

Methods

fromString :: String -> ConsumerGroupId

Generic ConsumerGroupId

Defined in Kafka.Consumer.Types

Associated Types

type Rep ConsumerGroupId :: Type -> Type

Methods

from :: ConsumerGroupId -> Rep ConsumerGroupId x

to :: Rep ConsumerGroupId x -> ConsumerGroupId

type Rep ConsumerGroupId

Defined in Kafka.Consumer.Types

type Rep ConsumerGroupId = D1 ('MetaData "ConsumerGroupId" "Kafka.Consumer.Types" "hw-kafka-client-4.0.3-inplace" 'True) (C1 ('MetaCons "ConsumerGroupId" 'PrefixI 'True) (S1 ('MetaSel ('Just "unConsumerGroupId") 'NoSourceUnpackedness 'NoSourceStrictness 'DecidedLazy) (Rec0 Text)))

kcKafkaPtr :: KafkaConsumer -> Kafka

kcKafkaConf :: KafkaConsumer -> KafkaConf

crMapKey :: (k -> k') -> ConsumerRecord k v -> ConsumerRecord k' v

Deprecated: Isn't concern of this library. Use first

crMapValue :: (v -> v') -> ConsumerRecord k v -> ConsumerRecord k v'

Deprecated: Isn't concern of this library. Use second

crMapKV :: (k -> k') -> (v -> v') -> ConsumerRecord k v -> ConsumerRecord k' v'

Deprecated: Isn't concern of this library. Use bimap

sequenceFirst :: (Bitraversable t, Applicative f) => t (f k) v -> f (t k v)

Deprecated: Isn't concern of this library. Use bitraverse id pure

traverseFirst :: (Bitraversable t, Applicative f) => (k -> f k') -> t k v -> f (t k' v)

Deprecated: Isn't concern of this library. Use bitraverse f pure

traverseFirstM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> t k v -> m (f (t k' v))

Deprecated: Isn't concern of this library. Use bitraverse id pure <$> bitraverse f pure r

traverseM :: (Traversable t, Applicative f, Monad m) => (v -> m (f v')) -> t v -> m (f (t v'))

Deprecated: Isn't concern of this library. Use sequenceA <$> traverse f r

bitraverseM :: (Bitraversable t, Applicative f, Monad m) => (k -> m (f k')) -> (v -> m (f v')) -> t k v -> m (f (t k' v'))

Deprecated: Isn't concern of this library. Use bisequenceA <$> bimapM f g r

data Subscription

A consumer subscription to a topic.

Examples

Typically you don't call the constructor directly, but combine settings:

consumerSub :: Subscription
consumerSub = topics [TopicName "kafka-client-example-topic"]
           <> offsetReset Earliest
           <> extraSubscriptionProps (fromList [("prop1", "value 1"), ("prop2", "value 2")])

Constructors

Subscription (Set TopicName) (Map Text Text)

Instances

Semigroup Subscription

Defined in Kafka.Consumer.Subscription

Methods

(<>) :: Subscription -> Subscription -> Subscription

sconcat :: NonEmpty Subscription -> Subscription

stimes :: Integral b => b -> Subscription -> Subscription

Monoid Subscription

Defined in Kafka.Consumer.Subscription

Methods

mempty :: Subscription

mappend :: Subscription -> Subscription -> Subscription

mconcat :: [Subscription] -> Subscription

topics :: [TopicName] -> Subscription

Build a subscription by giving the list of topic names only

offsetReset :: OffsetReset -> Subscription

Build a subscription by giving the offset reset parameter only

extraSubscriptionProps :: Map Text Text -> Subscription

Build a subscription by giving extra properties only

errorCallback :: (KafkaError -> String -> IO ()) -> Callback

Add a callback for errors.

Examples

Basic usage:

'setCallback' ('errorCallback' myErrorCallback)

myErrorCallback :: 'KafkaError' -> String -> IO ()
myErrorCallback kafkaError message = print $ show kafkaError <> "|" <> message

logCallback :: (KafkaLogLevel -> String -> String -> IO ()) -> Callback

Add a callback for logs.

Examples

Basic usage:

'setCallback' ('logCallback' myLogCallback)

myLogCallback :: 'KafkaLogLevel' -> String -> String -> IO ()
myLogCallback level facility message = print $ show level <> "|" <> facility <> "|" <> message

statsCallback :: (ByteString -> IO ()) -> Callback

Add a callback for stats. The passed ByteString contains an UTF-8 encoded JSON document and can e.g. be parsed using Data.Aeson.decodeStrict. For more information about the content of the JSON document see https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md.

Examples

Basic usage:

'setCallback' ('statsCallback' myStatsCallback)

myStatsCallback :: String -> IO ()
myStatsCallback stats = print $ show stats

rebalanceCallback :: (KafkaConsumer -> RebalanceEvent -> IO ()) -> Callback

Sets a callback that is called when rebalance is needed.

offsetCommitCallback :: (KafkaConsumer -> KafkaError -> [TopicPartition] -> IO ()) -> Callback

Sets a callback that is called when rebalance is needed.

The results of automatic or manual offset commits will be scheduled for this callback and is served by pollMessage.

If no partitions had valid offsets to commit this callback will be called with KafkaResponseError RdKafkaRespErrNoOffset which is not to be considered an error.

data ConsumerProperties

Properties to create KafkaConsumer.

Constructors

ConsumerProperties

Fields

• cpProps :: Map Text Text
• cpLogLevel :: Maybe KafkaLogLevel
• cpCallbacks :: [Callback]
• cpCallbackPollMode :: CallbackPollMode

Instances

Semigroup ConsumerProperties

Defined in Kafka.Consumer.ConsumerProperties

Methods

(<>) :: ConsumerProperties -> ConsumerProperties -> ConsumerProperties

sconcat :: NonEmpty ConsumerProperties -> ConsumerProperties

stimes :: Integral b => b -> ConsumerProperties -> ConsumerProperties

Monoid ConsumerProperties

Right biased so we prefer newer properties over older ones.

Defined in Kafka.Consumer.ConsumerProperties

Methods

mempty :: ConsumerProperties

mappend :: ConsumerProperties -> ConsumerProperties -> ConsumerProperties

mconcat :: [ConsumerProperties] -> ConsumerProperties

data CallbackPollMode

Whether the callback polling should be done synchronously or not.

Constructors

CallbackPollModeSync

You have to poll the consumer frequently to handle new messages as well as rebalance and keep alive events. This enables lowering the footprint and having full control over when polling happens, at the cost of manually managing those events.

CallbackPollModeAsync

Handle polling rebalance and keep alive events for you in a background thread.

Instances

Eq CallbackPollMode

Defined in Kafka.Consumer.ConsumerProperties

Methods

(==) :: CallbackPollMode -> CallbackPollMode -> Bool

(/=) :: CallbackPollMode -> CallbackPollMode -> Bool

Show CallbackPollMode

Defined in Kafka.Consumer.ConsumerProperties

Methods

showsPrec :: Int -> CallbackPollMode -> ShowS

show :: CallbackPollMode -> String

showList :: [CallbackPollMode] -> ShowS

brokersList :: [BrokerAddress] -> ConsumerProperties

Set the list of brokers to contact to connect to the Kafka cluster.

autoCommit :: Millis -> ConsumerProperties

Set the auto commit interval and enables auto commit.

noAutoCommit :: ConsumerProperties

Disable auto commit for the consumer.

noAutoOffsetStore :: ConsumerProperties

Disable auto offset store for the consumer.

See enable.auto.offset.store for more information.

groupId :: ConsumerGroupId -> ConsumerProperties

Set the consumer group id.

clientId :: ClientId -> ConsumerProperties

Set the consumer identifier.

setCallback :: Callback -> ConsumerProperties

Set the consumer callback.

For examples of use, see:

• errorCallback
• logCallback
• statsCallback

logLevel :: KafkaLogLevel -> ConsumerProperties

Set the logging level. Usually is used with debugOptions to configure which logs are needed.

compression :: KafkaCompressionCodec -> ConsumerProperties

Set the compression.codec for the consumer.

suppressDisconnectLogs :: ConsumerProperties

Suppresses consumer log.connection.close.

It might be useful to turn this off when interacting with brokers with an aggressive connection.max.idle.ms value.

statisticsInterval :: Millis -> ConsumerProperties

Set the statistics.interval.ms for the producer.

extraProps :: Map Text Text -> ConsumerProperties

Set any configuration options that are supported by librdkafka. The full list can be found here

extraProp :: Text -> Text -> ConsumerProperties

Set any configuration option that is supported by librdkafka. The full list can be found here

debugOptions :: [KafkaDebug] -> ConsumerProperties

Set debug features for the consumer. Usually is used with logLevel.

queuedMaxMessagesKBytes :: Int -> ConsumerProperties

Set queued.max.messages.kbytes

callbackPollMode :: CallbackPollMode -> ConsumerProperties

Set the callback poll mode. Default value is CallbackPollModeAsync.

runConsumer

Arguments

:: ConsumerProperties

-> Subscription

-> (KafkaConsumer -> IO (Either KafkaError a))

A callback function to poll and handle messages

-> IO (Either KafkaError a)

Deprecated: Use newConsumer/closeConsumer instead

Runs high-level kafka consumer. A callback provided is expected to call pollMessage when convenient.

newConsumer :: MonadIO m => ConsumerProperties -> Subscription -> m (Either KafkaError KafkaConsumer)

Create a KafkaConsumer. This consumer must be correctly released using closeConsumer.

assign :: MonadIO m => KafkaConsumer -> [TopicPartition] -> m (Maybe KafkaError)

Assigns the consumer to consume from the given topics, partitions, and offsets.

assignment :: MonadIO m => KafkaConsumer -> m (Either KafkaError (Map TopicName [PartitionId]))

Returns current consumer's assignment

subscription :: MonadIO m => KafkaConsumer -> m (Either KafkaError [(TopicName, SubscribedPartitions)])

Returns current consumer's subscription

pausePartitions :: MonadIO m => KafkaConsumer -> [(TopicName, PartitionId)] -> m KafkaError

Pauses specified partitions on the current consumer.

resumePartitions :: MonadIO m => KafkaConsumer -> [(TopicName, PartitionId)] -> m KafkaError

Resumes specified partitions on the current consumer.

committed :: MonadIO m => KafkaConsumer -> Timeout -> [(TopicName, PartitionId)] -> m (Either KafkaError [TopicPartition])

Retrieve committed offsets for topics+partitions.

position :: MonadIO m => KafkaConsumer -> [(TopicName, PartitionId)] -> m (Either KafkaError [TopicPartition])

Retrieve current positions (last consumed message offset+1) for the current running instance of the consumer. If the current consumer hasn't received any messages for a given partition, PartitionOffsetInvalid is returned.

seek :: MonadIO m => KafkaConsumer -> Timeout -> [TopicPartition] -> m (Maybe KafkaError)

Seek a particular offset for each provided TopicPartition

pollMessage

Arguments

:: MonadIO m

=> KafkaConsumer

-> Timeout

the timeout, in milliseconds

-> m (Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString)))

Left on error or timeout, right for success

Polls a single message

pollConsumerEvents :: KafkaConsumer -> Maybe Timeout -> IO ()

Polls the provided kafka consumer for events.

Events will cause application provided callbacks to be called.

The Timeout argument specifies the maximum amount of time (in milliseconds) that the call will block waiting for events.

This function is called on each pollMessage and, if runtime allows multi threading, it is called periodically in a separate thread to ensure the callbacks are handled ASAP.

There is no particular need to call this function manually unless some special cases in a single-threaded environment when polling for events on each pollMessage is not frequent enough.

pollMessageBatch :: MonadIO m => KafkaConsumer -> Timeout -> BatchSize -> m [Either KafkaError (ConsumerRecord (Maybe ByteString) (Maybe ByteString))]

Polls up to BatchSize messages. Unlike pollMessage this function does not return usual "timeout" errors. An empty batch is returned when there are no messages available.

This API is not available when CallbackPollMode is set to CallbackPollModeSync.

commitOffsetMessage :: MonadIO m => OffsetCommit -> KafkaConsumer -> ConsumerRecord k v -> m (Maybe KafkaError)

Commit message's offset on broker for the message's partition.

commitAllOffsets :: MonadIO m => OffsetCommit -> KafkaConsumer -> m (Maybe KafkaError)

Commit offsets for all currently assigned partitions.

commitPartitionsOffsets :: MonadIO m => OffsetCommit -> KafkaConsumer -> [TopicPartition] -> m (Maybe KafkaError)

Commit offsets for all currently assigned partitions.

storeOffsets :: MonadIO m => KafkaConsumer -> [TopicPartition] -> m (Maybe KafkaError)

Stores offsets locally

storeOffsetMessage :: MonadIO m => KafkaConsumer -> ConsumerRecord k v -> m (Maybe KafkaError)

Stores message's offset locally for the message's partition.

closeConsumer :: MonadIO m => KafkaConsumer -> m (Maybe KafkaError)

Closes the consumer.

See newConsumer

data RdKafkaRespErrT

Constructors

RdKafkaRespErrBegin

RdKafkaRespErrBadMsg

RdKafkaRespErrBadCompression

RdKafkaRespErrDestroy

RdKafkaRespErrFail

RdKafkaRespErrTransport

RdKafkaRespErrCritSysResource

RdKafkaRespErrResolve

RdKafkaRespErrMsgTimedOut

RdKafkaRespErrPartitionEof

RdKafkaRespErrUnknownPartition

RdKafkaRespErrFs

RdKafkaRespErrUnknownTopic

RdKafkaRespErrAllBrokersDown

RdKafkaRespErrInvalidArg

RdKafkaRespErrTimedOut

RdKafkaRespErrQueueFull

RdKafkaRespErrIsrInsuff

RdKafkaRespErrNodeUpdate

RdKafkaRespErrSsl

RdKafkaRespErrWaitCoord

RdKafkaRespErrUnknownGroup

RdKafkaRespErrInProgress

RdKafkaRespErrPrevInProgress

RdKafkaRespErrExistingSubscription

RdKafkaRespErrAssignPartitions

RdKafkaRespErrRevokePartitions

RdKafkaRespErrConflict

RdKafkaRespErrState

RdKafkaRespErrUnknownProtocol

RdKafkaRespErrNotImplemented

RdKafkaRespErrAuthentication

RdKafkaRespErrNoOffset

RdKafkaRespErrOutdated

RdKafkaRespErrTimedOutQueue

RdKafkaRespErrUnsupportedFeature

RdKafkaRespErrWaitCache

RdKafkaRespErrIntr

RdKafkaRespErrKeySerialization

RdKafkaRespErrValueSerialization

RdKafkaRespErrKeyDeserialization

RdKafkaRespErrValueDeserialization

RdKafkaRespErrPartial

RdKafkaRespErrReadOnly

RdKafkaRespErrNoent

RdKafkaRespErrUnderflow

RdKafkaRespErrInvalidType

RdKafkaRespErrRetry

RdKafkaRespErrPurgeQueue

RdKafkaRespErrPurgeInflight

RdKafkaRespErrFatal

RdKafkaRespErrInconsistent

RdKafkaRespErrGaplessGuarantee

RdKafkaRespErrMaxPollExceeded

RdKafkaRespErrUnknownBroker

RdKafkaRespErrNotConfigured

RdKafkaRespErrFenced

RdKafkaRespErrApplication

RdKafkaRespErrEnd

RdKafkaRespErrUnknown

RdKafkaRespErrNoError

RdKafkaRespErrOffsetOutOfRange

RdKafkaRespErrInvalidMsg

RdKafkaRespErrUnknownTopicOrPart

RdKafkaRespErrInvalidMsgSize

RdKafkaRespErrLeaderNotAvailable

RdKafkaRespErrNotLeaderForPartition

RdKafkaRespErrRequestTimedOut

RdKafkaRespErrBrokerNotAvailable

RdKafkaRespErrReplicaNotAvailable

RdKafkaRespErrMsgSizeTooLarge

RdKafkaRespErrStaleCtrlEpoch

RdKafkaRespErrOffsetMetadataTooLarge

RdKafkaRespErrNetworkException

RdKafkaRespErrCoordinatorLoadInProgress

RdKafkaRespErrCoordinatorNotAvailable

RdKafkaRespErrNotCoordinator

RdKafkaRespErrTopicException

RdKafkaRespErrRecordListTooLarge

RdKafkaRespErrNotEnoughReplicas

RdKafkaRespErrNotEnoughReplicasAfterAppend

RdKafkaRespErrInvalidRequiredAcks

RdKafkaRespErrIllegalGeneration

RdKafkaRespErrInconsistentGroupProtocol

RdKafkaRespErrInvalidGroupId

RdKafkaRespErrUnknownMemberId

RdKafkaRespErrInvalidSessionTimeout

RdKafkaRespErrRebalanceInProgress

RdKafkaRespErrInvalidCommitOffsetSize

RdKafkaRespErrTopicAuthorizationFailed

RdKafkaRespErrGroupAuthorizationFailed

RdKafkaRespErrClusterAuthorizationFailed

RdKafkaRespErrInvalidTimestamp

RdKafkaRespErrUnsupportedSaslMechanism

RdKafkaRespErrIllegalSaslState

RdKafkaRespErrUnsupportedVersion

RdKafkaRespErrTopicAlreadyExists

RdKafkaRespErrInvalidPartitions

RdKafkaRespErrInvalidReplicationFactor

RdKafkaRespErrInvalidReplicaAssignment

RdKafkaRespErrInvalidConfig

RdKafkaRespErrNotController

RdKafkaRespErrInvalidRequest

RdKafkaRespErrUnsupportedForMessageFormat

RdKafkaRespErrPolicyViolation

RdKafkaRespErrOutOfOrderSequenceNumber

RdKafkaRespErrDuplicateSequenceNumber

RdKafkaRespErrInvalidProducerEpoch

RdKafkaRespErrInvalidTxnState

RdKafkaRespErrInvalidProducerIdMapping

RdKafkaRespErrInvalidTransactionTimeout

RdKafkaRespErrConcurrentTransactions

RdKafkaRespErrTransactionCoordinatorFenced

RdKafkaRespErrTransactionalIdAuthorizationFailed

RdKafkaRespErrSecurityDisabled

RdKafkaRespErrOperationNotAttempted

RdKafkaRespErrKafkaStorageError

RdKafkaRespErrLogDirNotFound

RdKafkaRespErrSaslAuthenticationFailed

RdKafkaRespErrUnknownProducerId

RdKafkaRespErrReassignmentInProgress

RdKafkaRespErrDelegationTokenAuthDisabled

RdKafkaRespErrDelegationTokenNotFound

RdKafkaRespErrDelegationTokenOwnerMismatch

RdKafkaRespErrDelegationTokenRequestNotAllowed

RdKafkaRespErrDelegationTokenAuthorizationFailed

RdKafkaRespErrDelegationTokenExpired

RdKafkaRespErrInvalidPrincipalType

RdKafkaRespErrNonEmptyGroup

RdKafkaRespErrGroupIdNotFound

RdKafkaRespErrFetchSessionIdNotFound

RdKafkaRespErrInvalidFetchSessionEpoch

RdKafkaRespErrListenerNotFound

RdKafkaRespErrTopicDeletionDisabled

RdKafkaRespErrFencedLeaderEpoch

RdKafkaRespErrUnknownLeaderEpoch

RdKafkaRespErrUnsupportedCompressionType

RdKafkaRespErrStaleBrokerEpoch

RdKafkaRespErrOffsetNotAvailable

RdKafkaRespErrMemberIdRequired

RdKafkaRespErrPreferredLeaderNotAvailable

RdKafkaRespErrGroupMaxSizeReached

RdKafkaRespErrFencedInstanceId

RdKafkaRespErrEligibleLeadersNotAvailable

RdKafkaRespErrElectionNotNeeded

RdKafkaRespErrNoReassignmentInProgress

RdKafkaRespErrGroupSubscribedToTopic

RdKafkaRespErrInvalidRecord

RdKafkaRespErrUnstableOffsetCommit

RdKafkaRespErrEndAll

Instances

Bounded RdKafkaRespErrT

Defined in Kafka.Internal.RdKafka

Methods

minBound :: RdKafkaRespErrT

maxBound :: RdKafkaRespErrT

Enum RdKafkaRespErrT

Defined in Kafka.Internal.RdKafka

Methods

succ :: RdKafkaRespErrT -> RdKafkaRespErrT

pred :: RdKafkaRespErrT -> RdKafkaRespErrT

toEnum :: Int -> RdKafkaRespErrT

fromEnum :: RdKafkaRespErrT -> Int

enumFrom :: RdKafkaRespErrT -> [RdKafkaRespErrT]

enumFromThen :: RdKafkaRespErrT -> RdKafkaRespErrT -> [RdKafkaRespErrT]

enumFromTo :: RdKafkaRespErrT -> RdKafkaRespErrT -> [RdKafkaRespErrT]

enumFromThenTo :: RdKafkaRespErrT -> RdKafkaRespErrT -> RdKafkaRespErrT -> [RdKafkaRespErrT]

Eq RdKafkaRespErrT

Defined in Kafka.Internal.RdKafka

Methods

(==) :: RdKafkaRespErrT -> RdKafkaRespErrT -> Bool

(/=) :: RdKafkaRespErrT -> RdKafkaRespErrT -> Bool

Show RdKafkaRespErrT

Defined in Kafka.Internal.RdKafka

Methods

showsPrec :: Int -> RdKafkaRespErrT -> ShowS

show :: RdKafkaRespErrT -> String

showList :: [RdKafkaRespErrT] -> ShowS