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

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.

KafkaExpected KafkaExpectedResponse

Got a different form of a response than was requested.

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

Show KafkaClientError

data KafkaExpectedResponse

Type of response to expect, used for KafkaExpected error.

Constructors

ExpectedMetadata

ExpectedFetch

ExpectedProduce

Instances

Eq KafkaExpectedResponse

Show KafkaExpectedResponse

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

Show PartitionAndLeader

palTopic :: Lens' PartitionAndLeader TopicName

palPartition :: Lens' PartitionAndLeader Partition

palLeader :: Lens' PartitionAndLeader Leader

data TopicAndPartition

Constructors

TopicAndPartition

Fields

_tapTopic :: TopicName

_tapPartition :: Partition

Instances

Eq TopicAndPartition

Ord TopicAndPartition

Show TopicAndPartition

data TopicAndMessage

A topic with a serializable message.

Constructors

TopicAndMessage

Fields

_tamTopic :: TopicName

_tamMessage :: Message

Instances

Eq TopicAndMessage

Show TopicAndMessage

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.

makeRequest :: RequestMessage -> Kafka Request

Make a request, incrementing the _stateCorrelationId.

tryKafkaIO :: IO a -> Kafka a

Catch IOExceptions and wrap them in KafkaIOExceptions.

tryKafka :: Kafka a -> Kafka a

Catch IOExceptions and wrap them in KafkaIOExceptions.

doRequest :: Handle -> Request -> Kafka Response

runGetKafka :: Get a -> ByteString -> Kafka a

metadata :: MetadataRequest -> Kafka MetadataResponse

Send a metadata request

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 [PartitionAndLeader]

Find a leader and partition for the topic.

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

expectResponse :: KafkaExpectedResponse -> Getting (Leftmost b) ResponseMessage b -> Response -> Kafka b

Function to give an error when the response seems wrong.

protocolTime :: KafkaTime -> Time

Convert an abstract time to a serializable protocol value.

Fetching

ordinaryConsumerId :: ReplicaId

Default: -1

fetchRequest :: Offset -> Partition -> TopicName -> Kafka FetchRequest

Construct a fetch request from the values in the state.

fetch' :: Handle -> FetchRequest -> Kafka FetchResponse

Execute a fetch request and get the raw fetch response.

fetch :: FetchRequest -> Kafka FetchResponse

Execute a fetch request and get the raw fetch response. Round-robins the requests to addresses in the KafkaState.

fetchMessages :: FetchResponse -> [TopicAndMessage]

Extract out messages with their topics from a fetch response.

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)] -> RequestMessage

Create an offset request.