Kafka.Consumer.ConsumerProperties

Documentation

data ConsumerProperties

Properties to create KafkaConsumer.

Constructors

ConsumerProperties

Fields

• cpProps :: Map Text Text
• cpLogLevel :: Maybe KafkaLogLevel
• cpCallbacks :: [KafkaConf -> IO ()]
• 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

Constructors

CallbackPollModeSync

CallbackPollModeAsync

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

autoCommit :: Millis -> ConsumerProperties

noAutoCommit :: ConsumerProperties

Disables auto commit for the consumer

noAutoOffsetStore :: ConsumerProperties

Disables auto offset store for the consumer

groupId :: ConsumerGroupId -> ConsumerProperties

Consumer group id

clientId :: ClientId -> ConsumerProperties

setCallback :: (KafkaConf -> IO ()) -> ConsumerProperties

logLevel :: KafkaLogLevel -> ConsumerProperties

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

compression :: KafkaCompressionCodec -> ConsumerProperties

Sets the compression codec for the consumer.

suppressDisconnectLogs :: ConsumerProperties

Suppresses consumer disconnects logs.

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

extraProps :: Map Text Text -> ConsumerProperties

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

extraProp :: Text -> Text -> ConsumerProperties

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

debugOptions :: [KafkaDebug] -> ConsumerProperties

Sets debug features for the consumer. Usually is used with consumerLogLevel.

queuedMaxMessagesKBytes :: Int -> ConsumerProperties

callbackPollMode :: CallbackPollMode -> ConsumerProperties

Sets the callback poll mode.

The default CallbackPollModeAsync mode handles polling rebalance and keep alive events for you in a background thread.

With CallbacPollModeSync the user will poll the consumer frequently to handle new messages as well as rebalance and keep alive events. CallbacPollModeSync lets you can simplify hw-kafka-client's footprint and have full control over when polling happens at the cost of having to manage this yourself.

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

Sets a callback that is called when rebalance is needed.

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

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 KafkaError == KafkaResponseError RdKafkaRespErrNoOffset which is not to be considered an error.

errorCallback :: HasKafkaConf k => (KafkaError -> String -> IO ()) -> k -> IO ()

Add a callback for errors.

Examples

Basic usage:

setCallback (errorCallback myErrorCallback)

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

logCallback :: HasKafkaConf k => (KafkaLogLevel -> String -> String -> IO ()) -> k -> IO ()

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 :: HasKafkaConf k => (String -> IO ()) -> k -> IO ()

Add a callback for stats.

Examples

Basic usage:

setCallback (statsCallback myStatsCallback)

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