Kafka.Consumer

Documentation

runConsumer

Arguments

:: ConsumerProperties

-> Subscription

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

A callback function to poll and handle messages

-> IO (Either KafkaError a)

Runs high-level kafka consumer.

A callback provided is expected to call pollMessage when convenient.

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

Creates a kafka consumer. A new consumer MUST be closed with closeConsumer function.

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

Assigns specified partitions to a current consumer. Assigning an empty list means unassigning from all partitions that are currently assigned.

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

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)

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 the next message from a subscription

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.

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

Closes the consumer.

newtype TopicName

Topic name to be consumed

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 String

a simple topic name or a regex if started with ^

Instances

Eq TopicName

Methods

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

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

Ord TopicName

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

Methods

readsPrec :: Int -> ReadS TopicName

readList :: ReadS [TopicName]

readPrec :: ReadPrec TopicName

readListPrec :: ReadPrec [TopicName]

Show TopicName

Methods

showsPrec :: Int -> TopicName -> ShowS

show :: TopicName -> String

showList :: [TopicName] -> ShowS

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

Methods

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

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

Show OffsetCommit

Methods

showsPrec :: Int -> OffsetCommit -> ShowS

show :: OffsetCommit -> String

showList :: [OffsetCommit] -> ShowS

data PartitionOffset

Constructors

PartitionOffsetBeginning

PartitionOffsetEnd

PartitionOffset Int64

PartitionOffsetStored

PartitionOffsetInvalid

Instances

Eq PartitionOffset

Methods

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

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

Show PartitionOffset

Methods

showsPrec :: Int -> PartitionOffset -> ShowS

show :: PartitionOffset -> String

showList :: [PartitionOffset] -> ShowS

data TopicPartition

Kafka topic partition structure

Constructors

TopicPartition

Fields

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

Instances

Eq TopicPartition

Methods

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

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

Show TopicPartition

Methods

showsPrec :: Int -> TopicPartition -> ShowS

show :: TopicPartition -> String

showList :: [TopicPartition] -> ShowS

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
• crValue :: !v

Instances

Bifunctor ConsumerRecord

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

Bitraversable ConsumerRecord

Methods

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

Bifoldable ConsumerRecord

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

Functor (ConsumerRecord k)

Methods

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

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

Foldable (ConsumerRecord k)

Methods

fold :: Monoid m => ConsumerRecord k m -> 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)

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 v, Eq k) => Eq (ConsumerRecord k v)

Methods

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

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

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

Methods

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

readList :: ReadS [ConsumerRecord k v]

readPrec :: ReadPrec (ConsumerRecord k v)

readListPrec :: ReadPrec [ConsumerRecord k v]

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

Methods

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

show :: ConsumerRecord k v -> String

showList :: [ConsumerRecord k v] -> ShowS

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

RdKafkaRespErrEnd

RdKafkaRespErrUnknown

RdKafkaRespErrNoError

RdKafkaRespErrOffsetOutOfRange

RdKafkaRespErrInvalidMsg

RdKafkaRespErrUnknownTopicOrPart

RdKafkaRespErrInvalidMsgSize

RdKafkaRespErrLeaderNotAvailable

RdKafkaRespErrNotLeaderForPartition

RdKafkaRespErrRequestTimedOut

RdKafkaRespErrBrokerNotAvailable

RdKafkaRespErrReplicaNotAvailable

RdKafkaRespErrMsgSizeTooLarge

RdKafkaRespErrStaleCtrlEpoch

RdKafkaRespErrOffsetMetadataTooLarge

RdKafkaRespErrNetworkException

RdKafkaRespErrGroupLoadInProgress

RdKafkaRespErrGroupCoordinatorNotAvailable

RdKafkaRespErrNotCoordinatorForGroup

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

RdKafkaRespErrEndAll

Instances

Enum RdKafkaRespErrT

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

Methods

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

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

Show RdKafkaRespErrT

Methods

showsPrec :: Int -> RdKafkaRespErrT -> ShowS

show :: RdKafkaRespErrT -> String

showList :: [RdKafkaRespErrT] -> ShowS