fr33m0nk.alpakka-kafka.consumer — net.clojars.fr33m0nk/clj-alpakka-kafka 0.1.4

Akka Stream connector for subscribing to Kafka topics

->at-most-once-source clj

(->at-most-once-source consumer-settings topics)

Convenience for 'at-most once delivery' semantics. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#atMostOnceSourceK,V:akka.stream.javadsl.Source[org.apache.kafka.clients.consumer.ConsumerRecord[K,V],akka.kafka.javadsl.Consumer.Control]

->commit-with-metadata-partitioned-source clj

(->commit-with-metadata-partitioned-source consumer-settings
                                           topics
                                           consumer-record-metadata-extractor)

The same as #plainPartitionedSource but with offset commit with metadata support. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#commitWithMetadataPartitionedSourceK,V:akka.stream.javadsl.Source[akka.japi.Pair[org.apache.kafka.common.TopicPartition,akka.stream.javadsl.Source[akka.kafka.ConsumerMessage.CommittableMessage[K,V],akka.NotUsed]],akka.kafka.javadsl.Consumer.Control]

->commit-with-metadata-source clj

(->commit-with-metadata-source consumer-settings
                               topics
                               consumer-record-metadata-extractor)

The commitWithMetadataSource makes it possible to add additional metadata (in the form of a string) when an offset is committed based on the record. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#commitWithMetadataSourceK,V:akka.stream.javadsl.Source[akka.kafka.ConsumerMessage.CommittableMessage[K,V],akka.kafka.javadsl.Consumer.Control]

->committable-external-source clj

(->committable-external-source consumer
                               topic-partitions
                               group-id
                               commit-timeout-in-nanos)

The same as #plainExternalSource but with offset commit support. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#committableExternalSourceK,V:akka.stream.javadsl.Source[akka.kafka.ConsumerMessage.CommittableMessage[K,V],akka.kafka.javadsl.Consumer.Control]

->committable-partitioned-source clj

(->committable-partitioned-source consumer-settings topics)

The same as #plainPartitionedSource but with offset commit support. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#committablePartitionedSourceK,V:akka.stream.javadsl.Source[akka.japi.Pair[org.apache.kafka.common.TopicPartition,akka.stream.javadsl.Source[akka.kafka.ConsumerMessage.CommittableMessage[K,V],akka.NotUsed]],akka.kafka.javadsl.Consumer.Control]

->committable-source clj

(->committable-source consumer-settings topics)

The committableSource makes it possible to commit offset positions to Kafka. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#committableSourceK,V:akka.stream.javadsl.Source[akka.kafka.ConsumerMessage.CommittableMessage[K,V],akka.kafka.javadsl.Consumer.Control]

->plain-partitioned-source clj

(->plain-partitioned-source consumer-settings topics)

The plainPartitionedSource is a way to track automatic partition assignment from kafka. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#plainPartitionedSourceK,V:akka.stream.javadsl.Source[akka.japi.Pair[org.apache.kafka.common.TopicPartition,akka.stream.javadsl.Source[org.apache.kafka.clients.consumer.ConsumerRecord[K,V],akka.NotUsed]],akka.kafka.javadsl.Consumer.Control]

->plain-source clj

(->plain-source consumer-settings topics)

The plainSource emits ConsumerRecord elements (as received from the underlying KafkaConsumer). It has no support for committing offsets to Kafka. It can be used when the offset is stored externally or with auto-commit (note that auto-commit is by default disabled) https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#plainSourceK,V:akka.stream.javadsl.Source[org.apache.kafka.clients.consumer.ConsumerRecord[K,V],akka.kafka.javadsl.Consumer.Control]

consumer-settings clj

(consumer-settings actor-system
                   {:keys [group-id key-deserializer value-deserializer
                           bootstrap-servers auto-offset-reset
                           enable-auto-commit]
                    :or {auto-offset-reset "latest" enable-auto-commit "false"}
                    :as consumer-properties})

Settings for consumers. See akka.kafka.consumer section in reference.conf https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/ConsumerSettings.html

• Expects consumer-properties to be supplied with kebab-case-keyword keys Full config list can be found in org.apache.kafka.clients.consumer.ConsumerConfig

create-draining-control clj

Combine control and a stream completion signal materialized values into one, so that the stream can be stopped in a controlled way without losing commits https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#createDrainingControlT:akka.kafka.javadsl.Consumer.DrainingControl[T]

create-draining-control-with-pair clj

Combine control and a stream completion signal materialized values into one, so that the stream can be stopped in a controlled way without losing commits https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#createDrainingControlT:akka.kafka.javadsl.Consumer.DrainingControl[T]

create-noop-control clj

Combine control and a stream completion signal materialized values into one, so that the stream can be stopped in a controlled way without losing commits https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/javadsl/Consumer$.html#createNoopControl():akka.kafka.javadsl.Consumer.Control

IConsumerMessage clj protocol

committable-offset clj

(committable-offset committable-message)

ConsumerMessage is the Output element of committableSource. https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/ConsumerMessage$$CommittableMessage.html#committableOffset:akka.kafka.ConsumerMessage.CommittableOffset

consumer-record clj

(consumer-record committable-message)

Extracts consumer-record from an instance of CommittableMessage

partition-offset clj

(partition-offset transactional-message)

Output element of transactionalSource. The offset is automatically committed as by the Producer https://doc.akka.io/api/alpakka-kafka/4.0.2/akka/kafka/ConsumerMessage$$TransactionalMessage.html#partitionOffset:akka.kafka.ConsumerMessage.PartitionOffset

IConsumerRecord clj protocol

key clj

(key committable-message-or-consumer-record)

value clj

(value committable-message-or-consumer-record)

IControl clj protocol

Materialized value of the consumer Source

drain-and-shutdown clj

(drain-and-shutdown consumer-control stream-completion executor)

Stop producing messages from the Source, wait for stream completion and shut down the consumer Source so that all consumed messages reach the end of the stream. Failures in stream completion will be propagated, the source will be shut down anyway.

get-metrics clj

(get-metrics consumer-control)

Exposes underlying consumer or producer metrics (as reported by underlying Kafka client library)

shutdown clj

(shutdown consumer-control)

Shutdown the consumer Source. It will wait for outstanding offset commit requests before shutting down

shutdown? clj

(shutdown? consumer-control)

Shutdown status. The CompletionStage will be completed when the stage has been shut down and the underlying KafkaConsumer has been closed. Shutdown can be triggered from downstream cancellation, errors, or #shutdown.

stop clj

(stop consumer-control)

Stop producing messages from the Source. This does not stop underlying kafka consumer and does not unsubscribe from any topics/partitions.

topics->subscriptions clj

(topics->subscriptions topics)