Network.Kafka

Documentation

type KafkaAddress = (Host, Port)

data KafkaState

Constructors

KafkaState

Fields

• _stateName :: KafkaString

  Name to use as a client ID.

• _stateRequiredAcks :: RequiredAcks

  How many acknowledgements are required for producing.

• _stateRequestTimeout :: Timeout

  Time in milliseconds to wait for messages to be produced by broker.

• _stateWaitSize :: MinBytes

  Minimum size of response bytes to block for.

• _stateBufferSize :: MaxBytes

  Maximum size of response bytes to retrieve.

• _stateWaitTime :: MaxWaitTime

  Maximum time in milliseconds to wait for response.

• _stateCorrelationId :: CorrelationId

  An incrementing counter of requests.

• _stateBrokers :: Map Leader Broker

  Broker cache

• _stateConnections :: Map KafkaAddress (Pool Handle)

  Connection cache

• _stateTopicMetadata :: Map TopicName TopicMetadata

  Topic metadata cache

• _stateAddresses :: NonEmpty KafkaAddress

  Address cache

Instances

Show KafkaState

Methods

showsPrec :: Int -> KafkaState -> ShowS

show :: KafkaState -> String

showList :: [KafkaState] -> ShowS

stateWaitTime :: Lens' KafkaState MaxWaitTime

stateWaitSize :: Lens' KafkaState MinBytes

stateTopicMetadata :: Lens' KafkaState (Map TopicName TopicMetadata)

stateRequiredAcks :: Lens' KafkaState RequiredAcks

stateRequestTimeout :: Lens' KafkaState Timeout

stateName :: Lens' KafkaState KafkaString

stateCorrelationId :: Lens' KafkaState CorrelationId

stateConnections :: Lens' KafkaState (Map KafkaAddress (Pool Handle))

stateBufferSize :: Lens' KafkaState MaxBytes

stateBrokers :: Lens' KafkaState (Map Leader Broker)

stateAddresses :: Lens' KafkaState (NonEmpty KafkaAddress)

type Kafka = StateT KafkaState (ExceptT KafkaClientError IO)

The core Kafka monad.

type KafkaClientId = KafkaString

data KafkaClientError

Errors given from the Kafka monad.

Constructors

KafkaNoOffset

A response did not contain an offset.

KafkaDeserializationError String

A value could not be deserialized correctly.

KafkaInvalidBroker Leader

Could not find a cached broker for the found leader.

KafkaFailedToFetchMetadata

KafkaIOException IOException

Instances

Eq KafkaClientError

Methods

(==) :: KafkaClientError -> KafkaClientError -> Bool

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

Show KafkaClientError

Methods

showsPrec :: Int -> KafkaClientError -> ShowS

show :: KafkaClientError -> String

showList :: [KafkaClientError] -> ShowS

data KafkaTime

An abstract form of Kafka's time. Used for querying offsets.

Constructors

LatestTime

The latest time on the broker.

EarliestTime

The earliest time on the broker.

OtherTime Time

A specific time.

data PartitionAndLeader

Constructors

PartitionAndLeader

Fields

• _palTopic :: TopicName
• _palPartition :: Partition
• _palLeader :: Leader

Instances

Eq PartitionAndLeader

Methods

(==) :: PartitionAndLeader -> PartitionAndLeader -> Bool

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

Ord PartitionAndLeader

Methods

compare :: PartitionAndLeader -> PartitionAndLeader -> Ordering

(<) :: PartitionAndLeader -> PartitionAndLeader -> Bool

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

(>) :: PartitionAndLeader -> PartitionAndLeader -> Bool

(>=) :: PartitionAndLeader -> PartitionAndLeader -> Bool

max :: PartitionAndLeader -> PartitionAndLeader -> PartitionAndLeader

min :: PartitionAndLeader -> PartitionAndLeader -> PartitionAndLeader

Show PartitionAndLeader

Methods

showsPrec :: Int -> PartitionAndLeader -> ShowS

show :: PartitionAndLeader -> String

showList :: [PartitionAndLeader] -> ShowS

palTopic :: Lens' PartitionAndLeader TopicName

palPartition :: Lens' PartitionAndLeader Partition

palLeader :: Lens' PartitionAndLeader Leader

data TopicAndPartition

Constructors

TopicAndPartition

Fields

• _tapTopic :: TopicName
• _tapPartition :: Partition

Instances

Eq TopicAndPartition

Methods

(==) :: TopicAndPartition -> TopicAndPartition -> Bool

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

Ord TopicAndPartition

Methods

compare :: TopicAndPartition -> TopicAndPartition -> Ordering

(<) :: TopicAndPartition -> TopicAndPartition -> Bool

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

(>) :: TopicAndPartition -> TopicAndPartition -> Bool

(>=) :: TopicAndPartition -> TopicAndPartition -> Bool

max :: TopicAndPartition -> TopicAndPartition -> TopicAndPartition

min :: TopicAndPartition -> TopicAndPartition -> TopicAndPartition

Show TopicAndPartition

Methods

showsPrec :: Int -> TopicAndPartition -> ShowS

show :: TopicAndPartition -> String

showList :: [TopicAndPartition] -> ShowS

data TopicAndMessage

A topic with a serializable message.

Constructors

TopicAndMessage

Fields

• _tamTopic :: TopicName
• _tamMessage :: Message

Instances

Eq TopicAndMessage

Methods

(==) :: TopicAndMessage -> TopicAndMessage -> Bool

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

Show TopicAndMessage

Methods

showsPrec :: Int -> TopicAndMessage -> ShowS

show :: TopicAndMessage -> String

showList :: [TopicAndMessage] -> ShowS

tamTopic :: Lens' TopicAndMessage TopicName

tamMessage :: Lens' TopicAndMessage Message

tamPayload :: TopicAndMessage -> ByteString

Get the bytes from the Kafka message, ignoring the topic.

Configuration

defaultCorrelationId :: CorrelationId

Default: 0

defaultRequiredAcks :: RequiredAcks

Default: 1

defaultRequestTimeout :: Timeout

Default: 10000

defaultMinBytes :: MinBytes

Default: 0

defaultMaxBytes :: MaxBytes

Default: 1024 * 1024

defaultMaxWaitTime :: MaxWaitTime

Default: 0

mkKafkaState :: KafkaClientId -> KafkaAddress -> KafkaState

Create a consumer using default values.

addKafkaAddress :: KafkaAddress -> KafkaState -> KafkaState

runKafka :: KafkaState -> Kafka a -> IO (Either KafkaClientError a)

Run the underlying Kafka monad.

tryKafka :: Kafka a -> Kafka a

Catch IOExceptions and wrap them in KafkaIOExceptions.

makeRequest :: Handle -> ReqResp (Kafka a) -> Kafka a

Make a request, incrementing the _stateCorrelationId.

metadata :: MetadataRequest -> Kafka MetadataResponse

Send a metadata request to any broker.

metadata' :: Handle -> MetadataRequest -> Kafka MetadataResponse

Send a metadata request.

getTopicPartitionLeader :: TopicName -> Partition -> Kafka Broker

expect :: KafkaClientError -> (a -> Maybe b) -> a -> Kafka b

brokerPartitionInfo :: TopicName -> Kafka (Set PartitionAndLeader)

Find a leader and partition for the topic.

findMetadataOrElse :: [TopicName] -> Getting (Maybe a) KafkaState (Maybe a) -> KafkaClientError -> Kafka a

protocolTime :: KafkaTime -> Time

Convert an abstract time to a serializable protocol value.

updateMetadatas :: [TopicName] -> Kafka ()

updateMetadata :: TopicName -> Kafka ()

updateAllMetadata :: Kafka ()

withBrokerHandle :: Broker -> (Handle -> Kafka a) -> Kafka a

Execute a Kafka action with a Handle for the given Broker, updating the connections cache if needed.

When the action throws an IOException, it is caught and returned as a KafkaIOException in the Kafka monad.

Note that when the given action throws an exception, any state changes will be discarded. This includes both IOExceptions and exceptions thrown by throwError from Except.

withAddressHandle :: KafkaAddress -> (Handle -> Kafka a) -> Kafka a

Execute a Kafka action with a Handle for the given KafkaAddress, updating the connections cache if needed.

When the action throws an IOException, it is caught and returned as a KafkaIOException in the Kafka monad.

Note that when the given action throws an exception, any state changes will be discarded. This includes both IOExceptions and exceptions thrown by throwError from Except.

broker2address :: Broker -> KafkaAddress

withAnyHandle :: (Handle -> Kafka a) -> Kafka a

Like withAddressHandle, but round-robins the addresses in the KafkaState.

When the action throws an IOException, it is caught and returned as a KafkaIOException in the Kafka monad.

Note that when the given action throws an exception, any state changes will be discarded. This includes both IOExceptions and exceptions thrown by throwError from Except.

Offsets

data PartitionOffsetRequestInfo

Fields to construct an offset request, per topic and partition.

Constructors

PartitionOffsetRequestInfo

Fields

• _kafkaTime :: KafkaTime

  Time to find an offset for.

• _maxNumOffsets :: MaxNumberOfOffsets

  Number of offsets to retrieve.

getLastOffset :: KafkaTime -> Partition -> TopicName -> Kafka Offset

Get the first found offset.

getLastOffset' :: Handle -> KafkaTime -> Partition -> TopicName -> Kafka Offset

Get the first found offset.

offsetRequest :: [(TopicAndPartition, PartitionOffsetRequestInfo)] -> OffsetRequest

Create an offset request.