Kafka.Producer

Module to produce messages to Kafka topics.

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

import Control.Exception (bracket)
import Control.Monad (forM_)
import Data.ByteString (ByteString)
import Kafka.Producer

-- Global producer properties
producerProps :: ProducerProperties
producerProps = brokersList ["localhost:9092"]
             <> logLevel KafkaLogDebug

-- Topic to send messages to
targetTopic :: TopicName
targetTopic = TopicName "kafka-client-example-topic"

-- Run an example
runProducerExample :: IO ()
runProducerExample =
    bracket mkProducer clProducer runHandler >>= print
    where
      mkProducer = newProducer producerProps
      clProducer (Left _)     = pure ()
      clProducer (Right prod) = closeProducer prod
      runHandler (Left err)   = pure $ Left err
      runHandler (Right prod) = sendMessages prod

-- Example sending 2 messages and printing the response from Kafka
sendMessages :: KafkaProducer -> IO (Either KafkaError ())
sendMessages prod = do
  err1 <- produceMessage prod (mkMessage Nothing (Just "test from producer") )
  forM_ err1 print

  err2 <- produceMessage prod (mkMessage (Just "key") (Just "test from producer (with key)"))
  forM_ err2 print

  pure $ Right ()

mkMessage :: Maybe ByteString -> Maybe ByteString -> ProducerRecord
mkMessage k v = ProducerRecord
                  { prTopic = targetTopic
                  , prPartition = UnassignedPartition
                  , prKey = k
                  , prValue = v
                  }

Documentation

data KafkaProducer

The main type for Kafka message production, used e.g. to send messages.

Its constructor is intentionally not exposed, instead, one should used newProducer 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.2-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.2-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.2-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.2-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.2-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.2-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.2-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.2-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.2-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.2-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.2-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.2-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 DeliveryReport

The result of sending a message to the broker, useful for callbacks

Constructors

DeliverySuccess ProducerRecord Offset

The message was successfully sent at this offset

DeliveryFailure ProducerRecord KafkaError

The message could not be sent

NoMessageError KafkaError

An error occurred, but librdkafka did not attach any sent message

Instances

Eq DeliveryReport

Defined in Kafka.Producer.Types

Methods

(==) :: DeliveryReport -> DeliveryReport -> Bool

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

Show DeliveryReport

Defined in Kafka.Producer.Types

Methods

showsPrec :: Int -> DeliveryReport -> ShowS

show :: DeliveryReport -> String

showList :: [DeliveryReport] -> ShowS

Generic DeliveryReport

Defined in Kafka.Producer.Types

Associated Types

type Rep DeliveryReport :: Type -> Type

Methods

from :: DeliveryReport -> Rep DeliveryReport x

to :: Rep DeliveryReport x -> DeliveryReport

type Rep DeliveryReport

Defined in Kafka.Producer.Types

type Rep DeliveryReport = D1 (MetaData "DeliveryReport" "Kafka.Producer.Types" "hw-kafka-client-4.0.2-inplace" False) (C1 (MetaCons "DeliverySuccess" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 ProducerRecord) :*: S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Offset)) :+: (C1 (MetaCons "DeliveryFailure" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 ProducerRecord) :*: S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 KafkaError)) :+: C1 (MetaCons "NoMessageError" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 KafkaError))))

newtype ImmediateError

Data type representing an error that is caused by pre-flight conditions not being met

Constructors

ImmediateError KafkaError

Instances

Eq ImmediateError

Defined in Kafka.Producer.Types

Methods

(==) :: ImmediateError -> ImmediateError -> Bool

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

Show ImmediateError

Defined in Kafka.Producer.Types

Methods

showsPrec :: Int -> ImmediateError -> ShowS

show :: ImmediateError -> String

showList :: [ImmediateError] -> ShowS

data ProducePartition

Constructors

SpecifiedPartition !Int

The partition number of the topic

UnassignedPartition

Let the Kafka broker decide the partition

Instances

Eq ProducePartition

Defined in Kafka.Producer.Types

Methods

(==) :: ProducePartition -> ProducePartition -> Bool

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

Ord ProducePartition

Defined in Kafka.Producer.Types

Methods

compare :: ProducePartition -> ProducePartition -> Ordering

(<) :: ProducePartition -> ProducePartition -> Bool

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

(>) :: ProducePartition -> ProducePartition -> Bool

(>=) :: ProducePartition -> ProducePartition -> Bool

max :: ProducePartition -> ProducePartition -> ProducePartition

min :: ProducePartition -> ProducePartition -> ProducePartition

Show ProducePartition

Defined in Kafka.Producer.Types

Methods

showsPrec :: Int -> ProducePartition -> ShowS

show :: ProducePartition -> String

showList :: [ProducePartition] -> ShowS

Generic ProducePartition

Defined in Kafka.Producer.Types

Associated Types

type Rep ProducePartition :: Type -> Type

Methods

from :: ProducePartition -> Rep ProducePartition x

to :: Rep ProducePartition x -> ProducePartition

type Rep ProducePartition

Defined in Kafka.Producer.Types

type Rep ProducePartition = D1 (MetaData "ProducePartition" "Kafka.Producer.Types" "hw-kafka-client-4.0.2-inplace" False) (C1 (MetaCons "SpecifiedPartition" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) SourceUnpack SourceStrict DecidedStrict) (Rec0 Int)) :+: C1 (MetaCons "UnassignedPartition" PrefixI False) (U1 :: Type -> Type))

data ProducerRecord

Represents messages to be enqueued onto a Kafka broker (i.e. used for a producer)

Constructors

ProducerRecord

Fields

• prTopic :: !TopicName
• prPartition :: !ProducePartition
• prKey :: Maybe ByteString
• prValue :: Maybe ByteString

Instances

Eq ProducerRecord

Defined in Kafka.Producer.Types

Methods

(==) :: ProducerRecord -> ProducerRecord -> Bool

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

Show ProducerRecord

Defined in Kafka.Producer.Types

Methods

showsPrec :: Int -> ProducerRecord -> ShowS

show :: ProducerRecord -> String

showList :: [ProducerRecord] -> ShowS

Generic ProducerRecord

Defined in Kafka.Producer.Types

Associated Types

type Rep ProducerRecord :: Type -> Type

Methods

from :: ProducerRecord -> Rep ProducerRecord x

to :: Rep ProducerRecord x -> ProducerRecord

type Rep ProducerRecord

Defined in Kafka.Producer.Types

type Rep ProducerRecord = D1 (MetaData "ProducerRecord" "Kafka.Producer.Types" "hw-kafka-client-4.0.2-inplace" False) (C1 (MetaCons "ProducerRecord" PrefixI True) ((S1 (MetaSel (Just "prTopic") NoSourceUnpackedness SourceStrict DecidedStrict) (Rec0 TopicName) :*: S1 (MetaSel (Just "prPartition") NoSourceUnpackedness SourceStrict DecidedStrict) (Rec0 ProducePartition)) :*: (S1 (MetaSel (Just "prKey") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (Maybe ByteString)) :*: S1 (MetaSel (Just "prValue") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 (Maybe ByteString)))))

kpKafkaPtr :: KafkaProducer -> Kafka

kpKafkaConf :: KafkaProducer -> KafkaConf

kpTopicConf :: KafkaProducer -> TopicConf

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

deliveryCallback :: (DeliveryReport -> IO ()) -> Callback

Sets the callback for delivery reports.

/Note: A callback should not be a long-running process as it blocks librdkafka from continuing on the thread that handles the delivery callbacks. For callbacks to individual messsages see 'Kafka.Producer.produceMessage\''./

data ProducerProperties

Properties to create KafkaProducer.

Constructors

ProducerProperties

Fields

• ppKafkaProps :: Map Text Text
• ppTopicProps :: Map Text Text
• ppLogLevel :: Maybe KafkaLogLevel
• ppCallbacks :: [Callback]

Instances

Semigroup ProducerProperties

Defined in Kafka.Producer.ProducerProperties

Methods

(<>) :: ProducerProperties -> ProducerProperties -> ProducerProperties

sconcat :: NonEmpty ProducerProperties -> ProducerProperties

stimes :: Integral b => b -> ProducerProperties -> ProducerProperties

Monoid ProducerProperties

Right biased so we prefer newer properties over older ones.

Defined in Kafka.Producer.ProducerProperties

Methods

mempty :: ProducerProperties

mappend :: ProducerProperties -> ProducerProperties -> ProducerProperties

mconcat :: [ProducerProperties] -> ProducerProperties

brokersList :: [BrokerAddress] -> ProducerProperties

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

setCallback :: Callback -> ProducerProperties

Set the producer callback.

For examples of use, see:

• errorCallback
• logCallback
• statsCallback

logLevel :: KafkaLogLevel -> ProducerProperties

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

compression :: KafkaCompressionCodec -> ProducerProperties

Set the compression.codec for the producer.

topicCompression :: KafkaCompressionCodec -> ProducerProperties

Set the compression.codec for the topic.

sendTimeout :: Timeout -> ProducerProperties

Set the message.timeout.ms.

statisticsInterval :: Millis -> ProducerProperties

Set the statistics.interval.ms for the producer.

extraProps :: Map Text Text -> ProducerProperties

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

suppressDisconnectLogs :: ProducerProperties

Suppresses producer disconnects logs.

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

extraTopicProps :: Map Text Text -> ProducerProperties

Any *topic* configuration options that are supported by librdkafka. The full list can be found here

debugOptions :: [KafkaDebug] -> ProducerProperties

Sets debug features for the producer Usually is used with logLevel.

runProducer :: ProducerProperties -> (KafkaProducer -> IO (Either KafkaError a)) -> IO (Either KafkaError a)

Deprecated: Use 'newProducer'/'closeProducer' instead

Runs Kafka Producer. The callback provided is expected to call produceMessage or/and produceMessageBatch to send messages to Kafka.

newProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError KafkaProducer)

Creates a new kafka producer A newly created producer must be closed with closeProducer function.

produceMessage :: MonadIO m => KafkaProducer -> ProducerRecord -> m (Maybe KafkaError)

Sends a single message. Since librdkafka is backed by a queue, this function can return before messages are sent. See flushProducer to wait for queue to empty.

produceMessageBatch

Arguments

:: MonadIO m

=> KafkaProducer

-> [ProducerRecord]

-> m [(ProducerRecord, KafkaError)]

An empty list when the operation is successful, otherwise a list of "failed" messages with corresponsing errors.

Sends a batch of messages. Returns a list of messages which it was unable to send with corresponding errors. Since librdkafka is backed by a queue, this function can return before messages are sent. See flushProducer to wait for queue to empty.

produceMessage' :: MonadIO m => KafkaProducer -> ProducerRecord -> (DeliveryReport -> IO ()) -> m (Either ImmediateError ())

Sends a single message with a registered callback.

The callback can be a long running process, as it is forked by the thread that handles the delivery reports.

flushProducer :: MonadIO m => KafkaProducer -> m ()

Drains the outbound queue for a producer. This function is also called automatically when the producer is closed with closeProducer to ensure that all queued messages make it to Kafka.

closeProducer :: MonadIO m => KafkaProducer -> m ()

Closes the producer. Will wait until the outbound queue is drained before returning the control.

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