-- | Kafka adapter for the Shibuya queue processing framework.
--
-- This adapter integrates with Apache Kafka via
-- [kafka-effectful](https://github.com/shinzui/kafka-effectful) and
-- [hw-kafka-client](https://github.com/haskell-works/hw-kafka-client).
--
-- == Example Usage
--
-- @
-- import Shibuya.App (defaultAppConfig, runApp, mkProcessor)
-- import Shibuya.Adapter.Kafka (kafkaAdapter, defaultConfig)
-- import Kafka.Effectful.Consumer (runKafkaConsumer)
-- import Kafka.Consumer (brokersList, groupId, noAutoOffsetStore)
--
-- main :: IO ()
-- main = runEff
-- . runError \@KafkaError
-- . runKafkaConsumer props sub
-- $ do
-- adapter <- kafkaAdapter (defaultConfig [TopicName \"orders\"])
-- result <- runApp defaultAppConfig
-- [ (ProcessorId \"orders\", mkProcessor adapter myHandler)
-- ]
-- ...
-- @
--
-- == Message Lifecycle
--
-- 1. Messages are polled from Kafka in batches.
-- 2. Each message is wrapped as an @Ingested@ value with an @AckHandle@ via
-- Shibuya's adapter-facing smart constructors.
-- 3. On @AckOk@, the offset is stored locally; auto-commit or consumer close
-- later flushes stored offsets to the broker.
-- 4. On @AckRetry@, the offset is not stored. The adapter seeks the partition
-- back to the failed message so Kafka can redeliver it.
-- 5. On @AckDeadLetter@, the offset is stored after a loud stderr warning.
-- 6. On @AckHalt@, the partition is paused and offset is not stored.
--
-- == Serial Operation Required
--
-- This adapter must be run with serial message processing. librdkafka stores the
-- highest offset per partition without gap tracking, so concurrent finalization
-- can commit past an earlier message that failed, halted, or requested retry.
-- The 'Adapter' value does not contain the processor concurrency policy, so this
-- is a caller contract rather than a runtime guard: do not use @Async@ or @Ahead@
-- processing with this adapter until a gap-tracking commit layer exists.
--
-- == Dead Letters Are Dropped
--
-- This adapter does not include a DLQ producer. @AckDeadLetter@ stores the
-- message offset so the consumer group moves on, emits a warning to stderr, and
-- makes the message unrecoverable from that consumer group's committed position.
-- Core tracing still records the dead-letter decision and reason on the
-- per-message span.
--
-- Kafka does not expose a per-message delivery counter through this consumer
-- API, so 'Shibuya.Core.Types.Envelope.attempt' is always 'Nothing'. Handlers
-- cannot safely cap retries by counting attempts from the envelope; use an
-- external store, or return @AckHalt@ to stop the stream.
--
-- == Fatal Error Propagation
--
-- Non-fatal Kafka errors (poll timeouts, partition EOFs, and the rest of the
-- non-fatal set defined by @hw-kafka-streamly@'s 'Kafka.Streamly.Stream.isFatal')
-- are filtered out of the poll stream. Any error that survives that filter is
-- fatal by construction (for example, an SSL handshake failure, an authentication
-- failure, or an invalid broker configuration) and terminates the stream by
-- throwing through the 'Effectful.Error.Static.Error' @KafkaError@ effect. The
-- caller observes the failure by receiving a @Left err@ from the
-- @runError \@KafkaError@ scope around 'Shibuya.App.runApp'.
--
-- == AckHalt Partition Pause Semantics
--
-- @AckHalt@ pauses the originating partition by calling @pausePartitions@ from
-- @kafka-effectful@ and the processor stops. Polling therefore stops. After
-- @max.poll.interval.ms@ (librdkafka default: 300000 ms, or 5 minutes) the broker
-- may evict this consumer from its group and rebalance the partition to another
-- member, which resumes from the last committed offset. A single-member group
-- simply stalls until restart. Paused state is session-local and does not outlive
-- the current consumer.
--
-- == Rebalance Callback Helper
--
-- 'kafkaRebalanceHandler' is optional. Install it with
-- @Kafka.Consumer.setCallback (Kafka.Consumer.rebalanceCallback (kafkaRebalanceHandler state))@
-- before creating the consumer when you want stderr visibility into assignment
-- changes and eager cleanup of retry barriers for revoked partitions. Without it,
-- the seek barrier still self-heals when messages are finalized at or below the
-- barrier offset. Cooperative rebalance fencing of in-flight work is outside this
-- adapter's scope.
module Shibuya.Adapter.Kafka
( -- * Adapter
kafkaAdapter,
kafkaAdapterWith,
KafkaAdapterState,
newKafkaAdapterState,
kafkaRebalanceHandler,
-- * Configuration
KafkaAdapterConfig (..),
-- * Defaults
defaultConfig,
-- * Re-exports
TopicName (..),
BrokerAddress (..),
ConsumerGroupId (..),
OffsetCommit (..),
Timeout (..),
BatchSize (..),
KafkaError,
)
where
import Control.Concurrent.STM (atomically, writeTVar)
import Control.Monad.IO.Class (liftIO)
import Data.ByteString (ByteString)
import Data.IORef (atomicModifyIORef')
import Data.Map.Strict qualified as Map
import Data.Set qualified as Set
import Data.Text qualified as Text
import Effectful (Eff, IOE, (:>))
import Effectful.Error.Static (Error, catchError, throwError)
import Kafka.Consumer (RdKafkaRespErrT (..))
import Kafka.Consumer.Types (ConsumerGroupId (..), OffsetCommit (..), RebalanceEvent (..))
import Kafka.Consumer.Types qualified as KC
import Kafka.Effectful.Consumer.Effect (KafkaConsumer, commitAllOffsets, subscription)
import Kafka.Types (BatchSize (..), BrokerAddress (..), KafkaError (..), PartitionId, Timeout (..), TopicName (..))
import Shibuya.Adapter (Adapter (..))
import Shibuya.Adapter.Kafka.Config (KafkaAdapterConfig (..), defaultConfig)
import Shibuya.Adapter.Kafka.Internal (KafkaAdapterState (..), dropStaleRecords, ingestedStream, kafkaSource, mkIngested, newKafkaAdapterState, withConsumerLock)
import System.IO (hPutStrLn, stderr)
-- | Create a Kafka adapter with the given configuration.
--
-- The adapter operates within an existing 'KafkaConsumer' effect scope.
-- Consumer lifecycle (connection, group membership) is managed by
-- @runKafkaConsumer@ from kafka-effectful.
--
-- The adapter uses @noAutoOffsetStore@ with manual @storeOffsetMessage@ +
-- auto-commit for offset management. On shutdown, @commitAllOffsets@ flushes
-- offsets stored so far. Messages finalized during the drain window store offsets
-- after that explicit commit; let the surrounding @runKafkaConsumer@ scope end
-- normally after 'Shibuya.App.stopApp' returns so the consumer close path can
-- flush the final stored offsets under the same auto-commit mode.
--
-- The returned 'Shibuya.Adapter.Adapter.shutdown' action must be invoked while
-- the 'KafkaConsumer' effect is still in scope. Invoking it after
-- @runKafkaConsumer@ has returned will throw a 'KafkaError' from
-- @commitAllOffsets@ against a consumer that is no longer valid. This is a
-- caller-side invariant; the adapter does not catch the error.
kafkaAdapter ::
(KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>
KafkaAdapterConfig ->
Eff es (Adapter es (Maybe ByteString))
kafkaAdapter config = do
state <- liftIO newKafkaAdapterState
kafkaAdapterWith state config
-- | Create a Kafka adapter using caller-owned adapter state.
--
-- Use this when the same @KafkaAdapterState@ must also be referenced by
-- 'kafkaRebalanceHandler', which is installed in consumer properties before the
-- consumer is created.
kafkaAdapterWith ::
(KafkaConsumer :> es, Error KafkaError :> es, IOE :> es) =>
KafkaAdapterState ->
KafkaAdapterConfig ->
Eff es (Adapter es (Maybe ByteString))
kafkaAdapterWith state config = do
warnOnSubscriptionMismatch config
let messageSource =
ingestedStream (mkIngested state config) $
dropStaleRecords state $
kafkaSource state config
pure
Adapter
{ adapterName = "kafka:" <> Text.intercalate "," (map unTopicName config.topics),
source = messageSource,
shutdown = do
liftIO $ atomically $ writeTVar state.shutdownVar True
withConsumerLock state (commitAllOffsets OffsetCommit)
`catchError` \_ err -> case err of
KafkaResponseError RdKafkaRespErrNoOffset -> pure ()
_ -> throwError err
}
warnOnSubscriptionMismatch ::
(KafkaConsumer :> es, IOE :> es) =>
KafkaAdapterConfig ->
Eff es ()
warnOnSubscriptionMismatch config = do
liveSubscription <- subscription
let configured = Set.fromList config.topics
subscribed = Set.fromList (map fst liveSubscription)
if configured == subscribed
then pure ()
else
liftIO $
hPutStrLn stderr $
"[shibuya-kafka-adapter] WARNING: config topics differ from live Kafka subscription; configured="
<> show (Set.toList configured)
<> " subscribed="
<> show (Set.toList subscribed)
-- | Rebalance callback helper for caller-installed Kafka callbacks.
--
-- Install with 'Kafka.Consumer.setCallback' and
-- 'Kafka.Consumer.rebalanceCallback' before creating the consumer. The callback
-- logs every rebalance event to stderr and clears pending retry barriers for
-- revoked partitions. It does not fence in-flight work.
kafkaRebalanceHandler ::
KafkaAdapterState ->
KC.KafkaConsumer ->
RebalanceEvent ->
IO ()
kafkaRebalanceHandler state _consumer event = do
hPutStrLn stderr $ "[shibuya-kafka-adapter] rebalance: " <> show event
case event of
RebalanceRevoke revoked ->
clearRevokedBarriers revoked
_ ->
pure ()
where
clearRevokedBarriers :: [(TopicName, PartitionId)] -> IO ()
clearRevokedBarriers revoked =
atomicModifyIORef' state.seekBarrier $ \barriers ->
(foldr Map.delete barriers revoked, ())