Where communities thrive

  • Join over 1.5M+ people
  • Join over 100K+ communities
  • Free without limits
  • Create your own community
Repo info
    Is there any solution or configuration for suppressing "Kafka commit is to be retried" logs probably config or something else?
    Ryan Tomczik
    Hi everyone, I'm looking to commit offset batches with each event produced transactionally. The problem is it looks like you can only provide one PartitionOffset per event produced. My events are being created from several events consumed over several partitions, so I need to commit a batch of PartitionOffset per event. Is this possible?
    Vishal Bhavsar
    Hi, I'm looking to consume messages from earliest to a specific offset and then stop the consuming (thus stopping source from consuming/emitting more messages). What would be the best way to achieve this? I am using committablePartitionedSource so I have access to per-partition offset. How do I terminate the flow after a specific offset has been reached?
    Consumer.DrainingControl<Done> control =
        Consumer.committablePartitionedSource(consumerSettings, Subscriptions.topics(topic))
                pair -> {
                  Source<ConsumerMessage.CommittableMessage<String, String>, NotUsed> source =
                  return source
                      .map(message -> message.committableOffset())
                      .runWith(Committer.sink(committerSettings), system);
            .toMat(Sink.ignore(), Consumer::createDrainingControl)
    1 reply
    Hey, can someone help me, in stream i have combination of take and takeWithin, and i am wondering if takeWithin will start counter after it receives last message or first message that reaches takeWithin operator.
    Levi Ramsey
    I'm pretty sure takeWithin starts its timer at materialization (i.e. before it has seen a stream element)
    Burak Helvac─▒
    Please use kafka, but leave kafka-streams as soon as possible.
    Vishal Bhavsar
    Hi, how is the metadata from Consumer.commitWithMetadataPartitionedSource meant to be used? What does the param metadataFromRecord: Function[ConsumerRecord[K, V], String]) allow you to do? I can't find any examples. I want to stop processing when a message is greater than a given timestamp. I can see that the timestamp is available in the metadataFromRecord, but how can I use it in the result of commitWithMetadataPartitionedSource?
    3 replies
    Levi Ramsey

    The result of metadataFromRecord is only passed back to Kafka when committing an offset (see https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/OffsetAndMetadata.html). The message timestamp is available in any of the sources which give you a ConsumerRecord or a CommittableMessage without needing a commitWithMetadata source.

    For the other committable sources, you would call msg.record.timestamp to get the timestamp. So given StopAfterTimestamp, you could .takeWhile { msg => msg.record.timestamp <= StopAfterTimestamp }

    The only usecase for that metadata that I can see is if you have tooling which consumes the consumer-offset topic from Kafka (e.g. for observability) and you want to pass metadata like which hosts are committing offsets to that tooling
    Vishal Bhavsar
    That makes sense. Thank you for such a comprehensive response @leviramsey!
    Harry Tran
    Hello, does anyone have an example using Alpakka Kafka consumer in a Lagom application before? I have this in a class, and wire it in a LagomApplication, but I can only see the "starting" log, but not "executing" log (topic has messages being produced). I do not see the consumer group created when checking from broker side.
      private val consumerSettings = ConsumerSettings(actorSystem, new StringDeserializer, new StringDeserializer)
      logger.info("Starting the subscriber")
        .plainSource(consumerSettings, Subscriptions.topics(ExternalTopic))
        .mapAsync(1)(message => {
          val request = Json.parse(message.value).as[ExternalRequest]
          logger.info("Executing {}", request)
    1 reply
    Hm, what use have the CassandraWriteSettings? What's their purporse?
    1 reply
    Matthew de Detrich

    So I have an interesting problem where when I am subscribing to a Stream using Alpakka Kafka and right at the start of the stream I am using prefixAndTail(1).flatMapConcat to get the first element however it returns None even though topics are being sent to the Kafka topic. Interestingly I am not getting this problem with a local Kafka stream that I run with Docker.

    Does anyone know in what cases this occurs and also if prefixAndTail(1) is eager? i.e. will it wait for perpetuity until it happens to get an element or is there some kind of timeout?

    Matthew de Detrich
    So figured out the issue, turned out it was Main immediately terminating which was causing a shutdown.
    Is there any way to make sure that two messages aimed for two different topics either both end up in those topics or none of them does after sending them there using either Send Producer or any regular streaming producer?
    1 reply
    Dave Kichler
    Curious whether the consumption patterns for the Consumer sources are documented anywhere? I'm specifically curious about the semantics of Consumer.sourceWithOffsetContext when the source is assigned multiple partitions, how consumption is managed between partitions. I was under the impression the partitions were consumed from using round-robin distribution but cannot find documentation to back that up (or contradict/refute).
    3 replies
    Sean Kwak
    Can I ask how to do a conditional publish with msg data in the code shown in the following link?
    e.g. if msg.record.value contains some string, then publish otherwise skip etc.
    4 replies
    Ashish Sharma
    hey, what is the configuration for setting log.retention duration for a topic within client settings?
    Ashish Sharma
    I guess this has to be done at the time of topic creation from the client?
    Levi Ramsey
    Or done through the usual Kafka CLI tools (e.g. kafka-topics.sh)
    Koen Dejonghe
    Can I use HdfsFlow to write parquet files to hdfs? If so, how? Thank you.
    BTW, I have GenericRecords in my flow. I could use AvroParquetWriter, but that does not have the RotationStrategy and FilePathGenerator

    Hi all, I'm trying to use a dependency that adds Kinesis KPL support to akka. It has a KPLFlow class to provide that support. I'm relativly new to akka and flows, but my objective would be to have a kafka source, that is already created and have some type of sink, to replace the "native" kinesis sink and use this flow to deliver the records. Is there a way to extract from the flow class this? Or is it possible with just the flow "consume" from the kafka source and deliver to a target stream ?

    I've create a stackoverflow question regarding this https://stackoverflow.com/questions/73873966/akka-kafka-source-to-kinesis-sink-using-kpl


    Hi everyone. I have an application that receives an api request and relays its to a Kafka API Producer. Each request calls the producer to send a message to Kafka. The producer exists throughout the application lifetime and is shared for all requests.

    producer.send(new ProducerRecord[String, String](topic, requestBody))

    This works OK. Now I want to use instead, an alpakka Producer for the job. The code looks like this:

     val kafkaProducer = producerSettings.createKafkaProducer()
    val settingsWithProducer = producerSettings.withProducer(kafkaProducer)
    val done = Source.single(requestBody)
      .map(value => new ProducerRecord[String, String](topic, value))

    What are the advantages of the alpakka Producer over the plain, vanilla Producer? I don't know whether the new approach can help me handle a large number of API requests in order at the same time.

    Copeland Nicolas
    Thanks for sharing such great information, the post you published have some great information which is quite beneficial for me. I highly appreciated with your work abilities. https://www.mysainsburys.net/
    Ladinu Chandrasinghe
    Hello all, what is the proper way to implement batched at-most-once processing using alpakka-kafka?
    I have the following sub-optimal solution, but is there a better way that doesn't involve materializing an inner stream?
        def atMostOnceSource[K, V]: Source[ConsumerRecord[K, V], NotUsed] = {
             .committableSource[K, V](consumerSettings, Subscriptions.topics(allTopics))
             .groupedWithin(maxBatchSize, maxBatchDuration)
             .mapAsync(1) { messages: Seq[CommittableMessage[K, V]] =>
               val committableOffsetBatch =
                 .map(_ => messages)