Kafka.Consumer.ConsumerProperties

Documentation

data ConsumerProperties

Properties to create KafkaConsumer.

Constructors

ConsumerProperties

Fields

• cpProps :: Map String String
• cpLogLevel :: Maybe KafkaLogLevel
• callbacks :: [KafkaConf -> IO ()]

Instances

Semigroup ConsumerProperties

Right biased so we prefer newer properties over older ones.

Methods

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

sconcat :: NonEmpty ConsumerProperties -> ConsumerProperties

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

Monoid ConsumerProperties

Methods

mempty :: ConsumerProperties

mappend :: ConsumerProperties -> ConsumerProperties -> ConsumerProperties

mconcat :: [ConsumerProperties] -> ConsumerProperties

brokersList :: [BrokerAddress] -> ConsumerProperties

noAutoCommit :: ConsumerProperties

Disables auto commit 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 String String -> ConsumerProperties

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

extraProp :: String -> String -> 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

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

Sets a callback that is called when rebalance is needed.

Callback implementations suppose to watch for KafkaResponseError RdKafkaRespErrAssignPartitions and for KafkaResponseError RdKafkaRespErrRevokePartitions. Other error codes are not expected and would indicate something really bad happening in a system, or bugs in librdkafka itself.

A callback is expected to call assign according to the error code it receives.

• When RdKafkaRespErrAssignPartitions happens assign should be called with all the partitions it was called with. It is OK to alter partitions offsets before calling assign.
• When RdKafkaRespErrRevokePartitions happens assign should be called with an empty list of partitions.

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.

A callback is expected to call assign according to the error code it receives.

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.