Kafka.Conduit.Sink

Documentation

kafkaSink :: MonadResource m => ProducerProperties -> Sink ProducerRecord m (Maybe KafkaError)

Creates a kafka producer for given properties and returns a Sink.

This method of creating a Sink represents a simple case and does not provide access to KafkaProducer. For more complex scenarious kafkaSinkAutoClose or kafkaSinkNoClose can be used.

kafkaSinkAutoClose :: MonadResource m => KafkaProducer -> Sink ProducerRecord m (Maybe KafkaError)

Creates a Sink for a given KafkaProducer. The producer will be closed when the Sink is closed.

kafkaSinkNoClose :: MonadIO m => KafkaProducer -> Sink ProducerRecord m (Maybe KafkaError)

Creates a Sink for a given KafkaProducer. The producer will NOT be closed automatically.

kafkaBatchSinkNoClose :: MonadIO m => KafkaProducer -> Sink [ProducerRecord] m [(ProducerRecord, KafkaError)]

Creates a batching Sink for a given KafkaProducer. The producer will NOT be closed automatically.

commitOffsetsSink :: MonadIO m => KafkaConsumer -> Sink i m ()

Deprecated: Conceptually wrong thing to do. Does not require library support. Consider calling commitAllOffsets when appropriate.

Ignores incoming messages and commits offsets. Commit errors are ignored. This functionality should not exist as a Sink and will be removed in future versions. Consider having an effect instead:

mapMC (\_ -> commitAllOffsets OffsetCommit consumer)

flushThenCommitSink :: MonadIO m => KafkaConsumer -> KafkaProducer -> Sink i m ()

Deprecated: Conceptually wrong thing to do. Does not require library support. Consider calling 'flushProducer >>= commitAllOffsets' when appropriate.

Ignores incoming messages and commits offsets, but makes sure that producer has an empty outgoing queue. Commit errors are ignored. This functionality should not exist as a Sink and will be removed in future versions. Consider having an effect instead:

mapMC (\_ -> flushProducer producer >>= commitAllOffsets OffsetCommit consumer)