rdkafka-hs

Kafka bindings for Haskell backed by librdkafka, the C client for Apache Kafka.

rdkafka-hs is the maintained successor of hw-kafka-client, whose last Hackage release, 5.3.0, dates from 2023. It contains the full upstream history, the upstream commits that were never released, and fixes dependents needed, most importantly that a consumer fatal error is reported from every poll instead of being silently swallowed in the default callback poll mode. See CHANGELOG.md for everything that differs from hw-kafka-client 5.3.0.

Installation

rdkafka-hs is not on Hackage; depend on a release tag. In cabal.project:

source-repository-package
  type: git
  location: https://github.com/shinzui/rdkafka-hs.git
  tag: v1.1.0.0

and in each component's build-depends:

build-depends: rdkafka-hs >=1.1 && <1.2

The modules are Kafka.Consumer, Kafka.Producer, Kafka.Metadata, Kafka.Topic, Kafka.Transaction and Kafka.Version.

Migrating from hw-kafka-client

Replace any hw-kafka-client pin in cabal.project with the stanza above. Then, in every component that depends on it (libraries, executables, test suites and benchmarks), replace the dependency and its bounds; keeping >=5.3 && <6 would exclude rdkafka-hs's restarted version line:

-    hw-kafka-client >=5.3 && <6,
+    rdkafka-hs >=1.1 && <1.2,

Imports do not change: the Kafka.* module namespace, type names and record field names are the same. GHC 9.12 and librdkafka 2.15 are required (see Requirements). Behaviour worth knowing when you switch:

• A consumer fatal error (for example a static group member fenced by another consumer with the same group.instance.id) is now reported by every poll as Left (KafkaResponseError RdKafkaRespErrFatal), in both callback poll modes; consumerFatalError returns the cause. In CallbackPollModeAsync it used to be unobservable. A fatal consumer must be recreated.
• callbackPollMode is no longer reset by properties composed after it, so a consumer configured as callbackPollMode CallbackPollModeSync <> groupId g is now actually synchronous; under hw-kafka-client it ran in async mode. Code that reads cpCallbackPollMode gets a Maybe; use effectiveCallbackPollMode.
• flushProducer uses rd_kafka_flush; it still blocks until the queue is empty. flushProducerTimeout bounds the wait.
• In CallbackPollModeAsync, partitions given to assign now deliver their records to pollMessage.
• closeConsumer leaves the group gracefully but never frees the client, which waits for its garbage-collector finalizer; it can block against an unresponsive broker. destroyConsumer frees the client immediately, within a bound. Close first if you need a graceful departure.

Requirements

• GHC 9.12 or newer.
• librdkafka 2.15 or newer, with its headers. The build stops with rdkafka-hs requires librdkafka 2.15 or newer on an older header. Kafka.Version.librdkafkaVersion reports the library a program is actually linked against.
• c2hs, used to generate the FFI bindings at build time. Cabal builds it automatically through build-tool-depends.

Consumer

A high-level consumer joins a consumer group and has partitions assigned to it by the group; when consumers join or leave, the group rebalances the partitions among them. A working example is in ConsumerExample.hs. The examples connect to a broker at localhost:9092; build and run them with:

cabal run kafka-client-example -f examples

To select a broker, topic and consumer group, pass all three arguments. The program reads three input lines, produces four records and then consumes them:

cabal run kafka-client-example -f examples -- 127.0.0.1:39092 my-example-topic my-example-group

import Control.Exception (bracket)
import Control.Monad (replicateM_)
import Kafka.Consumer

-- Global consumer properties
consumerProps :: ConsumerProperties
consumerProps = brokersList ["localhost:9092"]
             <> groupId "consumer_example_group"
             <> noAutoCommit
             <> logLevel KafkaLogInfo

-- Subscription to topics
consumerSub :: Subscription
consumerSub = topics ["kafka-client-example-topic"]
           <> offsetReset Earliest

-- Running an example
runConsumerExample :: IO ()
runConsumerExample = do
    res <- bracket mkConsumer clConsumer runHandler
    print res
    where
      mkConsumer = newConsumer consumerProps consumerSub
      clConsumer (Left err) = return (Left err)
      clConsumer (Right kc) = maybe (Right ()) Left <$> closeConsumer kc
      runHandler (Left err) = return (Left err)
      runHandler (Right kc) = processMessages kc

-------------------------------------------------------------------
processMessages :: KafkaConsumer -> IO (Either KafkaError ())
processMessages kafka = do
    replicateM_ 10 $ do
      msg <- pollMessage kafka (Timeout 1000)
      putStrLn $ "Message: " <> show msg
      err <- commitAllOffsets OffsetCommit kafka
      putStrLn $ "Offsets: " <> maybe "Committed." show err
    return $ Right ()

Consumer topic names and metadata group queries reject embedded NUL. Partition IDs outside the native C integer range return KafkaBadSpecification through the operation’s existing error result before changing state. Other invalid inputs retain librdkafka’s validation and error conventions.

The examples in the sections below assume import Kafka.Consumer (or import Kafka.Producer) and the LambdaCase and OverloadedStrings extensions.

Callback poll modes

The default async mode requires an executable linked with -threaded. Without it, consumer construction returns KafkaBadSpecification; select callbackPollMode CallbackPollModeSync to use a nonthreaded executable.

librdkafka reports rebalances, offset commits, errors, logs and statistics through callbacks, which run only while something polls the client. A consumer has one of two callback poll modes, set with callbackPollMode:

• CallbackPollModeAsync (the default): a background thread polls every 100 ms to serve callbacks, and records are redirected to a separate queue that pollMessage reads. Callbacks keep running while the application is busy, so librdkafka's max.poll.interval.ms cannot detect an application that has stopped polling.
• CallbackPollModeSync: callbacks run inside the application's own pollMessage calls, and nothing runs between them. pollMessageBatch is not available in this mode.

syncProps :: ConsumerProperties
syncProps = brokersList ["localhost:9092"]
         <> groupId "my-group"
         <> callbackPollMode CallbackPollModeSync

Fatal errors

Some errors are permanent: once librdkafka raises a fatal error on a consumer, it stops all group activity, and the only remedy is a new consumer. Every poll then returns Left (KafkaResponseError RdKafkaRespErrFatal), in both poll modes, and consumerFatalError returns the underlying cause:

consumeLoop :: KafkaConsumer -> IO ()
consumeLoop consumer =
  pollMessage consumer (Timeout 1000) >>= \case
    Right record -> print (crValue record) >> consumeLoop consumer
    Left (KafkaResponseError RdKafkaRespErrTimedOut) -> consumeLoop consumer
    Left (KafkaResponseError RdKafkaRespErrFatal) -> do
      cause <- consumerFatalError consumer
      putStrLn ("consumer is permanently broken: " <> show cause)
      -- close or destroy it, then create a new consumer
    Left err -> print err >> consumeLoop consumer

Waiting for records without polling

setConsumerQueueIOEvent makes librdkafka write a payload to a non-blocking file descriptor when records or events reach the queue pollMessage reads, so an application can wait with threadWaitRead instead of polling on a timer. The notification is edge-triggered: one wake-up can stand for several records, or for an event that is not a record. Discard pending wake-ups, poll the queue empty, then wait, and bound the wait so a fatal error is still noticed:

import Control.Concurrent (threadWaitRead)
import Control.Exception (bracket, try)
import Control.Monad (when)
import Data.ByteString (ByteString)
import Foreign.Marshal.Alloc (allocaBytes)
import System.Posix.IO (FdOption (..), closeFd, createPipe, fdReadBuf, setFdOption)
import System.Posix.Types (ByteCount, Fd)
import System.Timeout (timeout)

-- | Handle records as librdkafka announces them, waking at least once a second.
-- Returns when the consumer turns fatal.
waitAndConsume :: KafkaConsumer -> (ConsumerRecord (Maybe ByteString) (Maybe ByteString) -> IO ()) -> IO ()
waitAndConsume consumer handle =
  bracket createPipe release $ \(readEnd, writeEnd) -> do
    -- NonBlockingRead sets O_NONBLOCK, which librdkafka's writes need too.
    setFdOption readEnd NonBlockingRead True
    setFdOption writeEnd NonBlockingRead True
    setConsumerQueueIOEvent consumer (Just (writeEnd, "1"))
    let drain =
          pollMessage consumer (Timeout 0) >>= \case
            Right record -> handle record >> drain
            Left (KafkaResponseError RdKafkaRespErrTimedOut) -> pure True -- queue empty
            Left (KafkaResponseError RdKafkaRespErrFatal) -> pure False
            Left _ -> drain -- a non-record event
        loop = do
          discardWakeups readEnd -- before draining, so no wake-up is lost
          alive <- drain
          when alive $ do
            _ <- timeout 1000000 (threadWaitRead readEnd)
            loop
    loop
  where
    release (readEnd, writeEnd) = do
      setConsumerQueueIOEvent consumer Nothing -- before closing the descriptor
      closeFd readEnd
      closeFd writeEnd

discardWakeups :: Fd -> IO ()
discardWakeups fd =
  allocaBytes 64 $ \buf ->
    (try (fdReadBuf fd buf 64) :: IO (Either IOError ByteCount)) >>= \case
      Right n | n > 0 -> discardWakeups fd
      _ -> pure ()

Reading partitions without joining a group

newAssignedConsumer creates a consumer that never subscribes, so it never joins its consumer group: it reads only the partitions given to assign, from the offsets given there. With enable.partition.eof set, librdkafka reports when a partition's end is reached, and pollMessageAttributed says which partition and offset each such error belongs to. Together they make a read-only scanner:

-- | Print partition 0 of "events" from its first offset to its current end.
scanPartition :: IO ()
scanPartition =
  bracket (newAssignedConsumer props) (either (const (pure ())) destroyConsumer) $ \case
    Left err -> print err
    Right consumer -> do
      _ <- assign consumer [TopicPartition "events" (PartitionId 0) PartitionOffsetBeginning]
      let loop =
            pollMessageAttributed consumer (Timeout 1000) >>= \case
              Right record -> print (crOffset record, crValue record) >> loop
              Left ConsumerMessageError {cmeError = KafkaResponseError RdKafkaRespErrPartitionEof, cmeOffset} ->
                putStrLn ("reached the end at " <> show cmeOffset)
              Left ConsumerMessageError {cmeError = KafkaResponseError RdKafkaRespErrTimedOut} -> loop
              Left err -> print err
      loop
  where
    props = brokersList ["localhost:9092"]
         <> groupId "events-observer" -- required by librdkafka; never joined
         <> noAutoCommit
         <> extraProp "enable.partition.eof" "true"

Shutting down a consumer

closeConsumer is the graceful close: it leaves the consumer group (letting the group rebalance at once) and commits offsets if auto-commit is enabled. It does not free the librdkafka client, which otherwise waits for a garbage collection, and it can block against an unresponsive broker. destroyConsumer stops the consumer's threads and closes its broker connections before it returns; every other operation on the consumer is invalid afterwards. Closing and destroying are each safe to repeat.

shutdownConsumer :: KafkaConsumer -> IO ()
shutdownConsumer consumer = do
  _ <- closeConsumer consumer -- leave the group; commits if auto-commit is on
  destroyConsumer consumer    -- free the client and close its connections now

To bound shutdown when the broker may not answer, call only destroyConsumer; the group then notices the departure when the consumer's session times out.

Producer

The producer sends messages to any number of topics; the target topic is part of each ProducerRecord given to produceMessage. A working example is in ProducerExample.hs.

{-# LANGUAGE OverloadedStrings #-}
import Control.Exception (bracket)
import Control.Monad (forM_)
import Data.ByteString (ByteString)
import Kafka.Producer

-- Global producer properties
producerProps :: ProducerProperties
producerProps = brokersList ["localhost:9092"]
             <> logLevel KafkaLogDebug

-- Topic to send messages to
targetTopic :: TopicName
targetTopic = "kafka-client-example-topic"

-- Run an example
runProducerExample :: IO ()
runProducerExample =
    bracket mkProducer clProducer runHandler >>= print
    where
      mkProducer = newProducer producerProps
      clProducer (Left _)     = return ()
      clProducer (Right prod) = closeProducer prod
      runHandler (Left err)   = return $ Left err
      runHandler (Right prod) = sendMessages prod

sendMessages :: KafkaProducer -> IO (Either KafkaError ())
sendMessages prod = do
  err1 <- produceMessage prod (mkMessage Nothing (Just "test from producer") )
  forM_ err1 print

  err2 <- produceMessage prod (mkMessage (Just "key") (Just "test from producer (with key)"))
  forM_ err2 print

  return $ Right ()

mkMessage :: Maybe ByteString -> Maybe ByteString -> ProducerRecord
mkMessage k v = ProducerRecord
                  { prTopic = targetTopic
                  , prPartition = UnassignedPartition
                  , prKey = k
                  , prValue = v
                  , prHeaders = mempty
                  }

Delivery reports

The producer keeps its own queue of outgoing messages. produceMessage returning means the message is queued, not that it has reached Kafka. A message that cannot be sent within message.timeout.ms (see librdkafka's configuration) is dropped from the queue and reported as failed. sendTimeout (Timeout 0) makes the timeout infinite, so messages are never dropped:

producerProps :: ProducerProperties
producerProps = brokersList ["localhost:9092"]
             <> sendTimeout (Timeout 0) -- for librdkafka "0" means "infinite"

deliveryCallback receives a DeliveryReport for every message: delivered at an offset, failed with an error, or an error librdkafka could not attach to a message. Callbacks run while the producer is polled, for example by flushProducer:

producerProps :: ProducerProperties
producerProps = brokersList ["localhost:9092"]
             <> setCallback (deliveryCallback report)
  where
    report = \case
      DeliverySuccess record offset -> putStrLn ("delivered to " <> show (prTopic record) <> " at " <> show offset)
      DeliveryFailure _ err -> putStrLn ("not delivered: " <> show err)
      NoMessageError err -> putStrLn ("delivery error: " <> show err)

Synchronous sending of messages

librdkafka has no synchronous produce call, but produceMessage' takes a callback for the message's own delivery report, which an MVar turns into a synchronous send:

{-# LANGUAGE LambdaCase #-}
import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
import Control.Monad.IO.Class (MonadIO (liftIO))
import Kafka.Consumer (Offset)
import Kafka.Producer

sendMessageSync :: MonadIO m => KafkaProducer -> ProducerRecord -> m (Either KafkaError Offset)
sendMessageSync producer record = liftIO $ do
  var <- newEmptyMVar
  res <- produceMessage' producer record (putMVar var)
  case res of
    Left (ImmediateError err) -> pure (Left err)
    Right () -> do
      -- Flush so the delivery report is served rather than waited for forever.
      flushProducer producer
      takeMVar var >>= \case
        DeliverySuccess _ offset -> pure (Right offset)
        DeliveryFailure _ err -> pure (Left err)
        NoMessageError err -> pure (Left err)

flushProducer waits until the queue is empty, which against an unreachable broker can take as long as message.timeout.ms.

Bounded shutdown

closeProducer flushes the queue without a time limit. To bound shutdown, flush with flushProducerTimeout and purge whatever is left; purged messages are reported to the delivery callback as failed (RdKafkaRespErrPurgeQueue or RdKafkaRespErrPurgeInflight):

shutdownProducer :: KafkaProducer -> IO ()
shutdownProducer producer =
  flushProducerTimeout producer (Timeout 5000) >>= \case
    Nothing -> pure ()
    Just _ -> do
      _ <- purgeProducer producer [PurgeQueue, PurgeInFlight]
      _ <- flushProducerTimeout producer (Timeout 1000) -- serves the purged reports
      pure ()

Development

The Nix flake provides GHC, HLS, cabal, c2hs and librdkafka. Enter the shell with nix develop (or let direnv load it), then:

just build            # cabal build all
just test             # cabal test rdkafka-hs-test (unit tests, no broker)
just test-it          # integration tests against the private broker (below)
just build-examples   # build the example executable
just fmt              # nix fmt: fourmolu, cabal-gild and nixpkgs-fmt
just check            # sandboxed build and unit tests, formatting, docs and bindings checks
just docs-check       # validate docs/adr and docs/improvement-requests
just bindings-check   # verify docs/bindings.md against the binding module

nix build builds the package and runs its unit tests in the Nix sandbox.

docs/bindings.md lists every function librdkafka's public header exports and which of them this package binds. It is generated from the header of the librdkafka release that pkg-config finds (cloned once into ~/.cache/rdkafka-hs/, or named by RDKAFKA_SOURCE_HEADER) and from src/Kafka/Internal/RdKafka.chs; regenerate it with just bindings-inventory after changing the bindings.

Design decisions are recorded in docs/adr/.

Integration tests

The current functional candidate passes 78 unit and 753 integration cases in both normal and Haskell coverage runs on macOS 26.7/arm64, GHC 9.12.4, librdkafka 2.15.0 and private Redpanda 26.2.1. Four additional runtime checks run in an actual nonthreaded executable.

A clean source archive also builds the library, examples and suites on Ubuntu 24.04.5/aarch64 with GHC 9.14.1, librdkafka 2.16.0 and Apache Kafka 4.4.0. That tuple passes 78 unit, four nonthreaded and 279 native-lifetime cases, nine distributed leader/group/outage/transaction cases, and seventeen SASL PLAIN/TLS/ACL/credential-reconnect cases. Linux diagnostics detect deliberate invalid-access and native-header retention controls; complementary Darwin ASan covers invalid access with leak detection disabled. These are named scenario results, with diagnostic limits recorded in the compatibility matrix.

Comprehensive qualification remains open: two producer exception-path coverage ticks and the final all-evidence/package gate are tracked by the follow-up plan. The historical primary release record retains its own exact source and package results.

The integration suite rdkafka-hs-integration needs a running Kafka broker and is disabled unless the manual integration flag is set, so cabal build all, cabal test and nix build never need one.

The suite runs against a private Redpanda broker owned by this project. It never uses the machine-wide broker at 127.0.0.1:9092, and refuses to start if RDKAFKA_TEST_BROKER names it, because the tests create and delete topics.

Requirements: an Apple Silicon Mac on macOS 26 or newer with Apple Container installed and its service running (container system start).

just test-it          # start the broker if needed, then run the suite
just redpanda-up      # start the broker (idempotent)
just redpanda-status  # exits non-zero unless the broker is serving
just redpanda-logs    # tail the broker logs
just redpanda-down    # stop the broker, keeping its data
just redpanda-purge   # delete its containers, network and volume

Arguments after just test-it go to cabal; to run one group, for example:

just test-it --test-option=-p '--test-option=/queue event/'

process-compose up also starts the broker. The dev shell exports RDKAFKA_TEST_BROKER=127.0.0.1:39092 unless it is already set. Every topic and consumer group a run creates carries a random per-run prefix, so runs do not interfere with each other.

The ports are the redpanda.*-port variables in .seihou/manifest.json, rendered into nix/redpanda.nix by seihou run nix-haskell-flake; changing one means changing it there, in flake.module.nix and in test-integration/Kafka/TestEnv.hs.

Credits

rdkafka-hs is a fork of hw-kafka-client by Alexey Raga and its contributors, which was itself inspired by Haskakafka.