Kafka.Producer.Sync

This module provides a synchronous interface on top of the hw-kafka-client

It works by using MVars managed in two different queues. Each request is sent as soon as there are no other effectively equal Kafka records in-flight. This is done in order to make sure that there is no ambiguity as to which MVar to resolve.

Currently, this implements fair sending. For all requests, the oldest pending request should be sent first.

Sync producer

data SyncKafkaProducer

A producer for sending messages to Kafka and waiting for the DeliveryReport

newSyncProducer :: MonadIO m => ProducerProperties -> m (Either KafkaError SyncKafkaProducer)

Create a new SyncKafkaProducer

Note: since this library wraps the regular hw-kafka-client, please be aware that you should not set the delivery report callback. As it is set internally.

produceRecord :: MonadIO m => SyncKafkaProducer -> ProducerRecord -> m (Either KafkaError ())

Re-exports

Record datatypes

data ProducerRecord

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" 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)))))

newtype TopicName

Constructors

TopicName

Fields

• unTopicName :: Text

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

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" True) (C1 (MetaCons "TopicName" PrefixI True) (S1 (MetaSel (Just "unTopicName") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Text)))

data ProducePartition

Constructors

SpecifiedPartition !Int

UnassignedPartition

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" False) (C1 (MetaCons "SpecifiedPartition" PrefixI False) (S1 (MetaSel (Nothing :: Maybe Symbol) SourceUnpack SourceStrict DecidedUnpack) (Rec0 Int)) :+: C1 (MetaCons "UnassignedPartition" PrefixI False) (U1 :: Type -> Type))

Errors

data KafkaError

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" 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))))

Producer configuration

data ProducerProperties

Constructors

ProducerProperties

Fields

• ppKafkaProps :: Map Text Text
• ppTopicProps :: Map Text Text
• ppLogLevel :: Maybe KafkaLogLevel
• ppCallbacks :: [KafkaConf -> IO ()]

Instances

Semigroup ProducerProperties

Defined in Kafka.Producer.ProducerProperties

Methods

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

sconcat :: NonEmpty ProducerProperties -> ProducerProperties

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

Monoid ProducerProperties

Defined in Kafka.Producer.ProducerProperties

Methods

mempty :: ProducerProperties

mappend :: ProducerProperties -> ProducerProperties -> ProducerProperties

mconcat :: [ProducerProperties] -> ProducerProperties

Configuration helpers

brokersList :: [BrokerAddress] -> ProducerProperties

Set brokers for producer

logLevel :: KafkaLogLevel -> ProducerProperties

Set log-level for producer

compression :: KafkaCompressionCodec -> ProducerProperties

Set compression level for producer

topicCompression :: KafkaCompressionCodec -> ProducerProperties

Set topic compression for producer

sendTimeout :: Timeout -> ProducerProperties

Set send timeout for producer

extraProps :: Map Text Text -> ProducerProperties

Set extra properties for producer

suppressDisconnectLogs :: ProducerProperties

Suppress disconnect log lines

extraTopicProps :: Map Text Text -> ProducerProperties

Configure extra topic properties

debugOptions :: [KafkaDebug] -> ProducerProperties

Add KafkaDebug options

Other datatypes

newtype BrokerAddress

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

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" True) (C1 (MetaCons "BrokerAddress" PrefixI True) (S1 (MetaSel (Just "unBrokerAddress") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Text)))

data KafkaCompressionCodec

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" 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

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" 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 KafkaLogLevel

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

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-3.0.0-f3f0ff4ee651ea9a164f922cc2bc7402c8529ca1a9367543bbc14e3ba60051d1" True) (C1 (MetaCons "Timeout" PrefixI True) (S1 (MetaSel (Just "unTimeout") NoSourceUnpackedness NoSourceStrictness DecidedLazy) (Rec0 Int)))