Where communities thrive

  • Join over 1.5M+ people
  • Join over 100K+ communities
  • Free without limits
  • Create your own community
Repo info
    Joe Zulli

    Hi all! I have a newbie question about working with Sinks in conjunction with ProducerMessage.multi. Essentially, I have a source that is giving me multi-line strings, and I want to send each line as a separate ProducerRecord into Kafka. Something like this:

        val done: Future[Done] =
          Source(1 to 100)
            .map(num => (0 to num).toList )
            .map(list => ProducerMessage.multi[Nothing, String](list.map(new ProducerRecord("topic", _)).toSeq))

    The error I get is:

    type mismatch;
     found   : akka.stream.scaladsl.Sink[org.apache.kafka.clients.producer.ProducerRecord[String,String],scala.concurrent.Future[akka.Done]]
     required: akka.stream.Graph[akka.stream.SinkShape[akka.kafka.ProducerMessage.Envelope[Nothing,String,akka.NotUsed]],scala.concurrent.Future[akka.Done]]

    Not sure how to make all of the types happy. If anyone can point me in the right direction, it would be much appreciated!

    3 replies
    Hi, when using Producer-Multi-Messages, is it normal that the commit offsets are incremented by the number of elements within the multi message? I'm sending two things to two separate topics with the help of a multi message, and the commit offset increases by 2 on every topic.
    1 reply
    Using alpakka, is it possible to time out and release a consumer if no message was received for e.g. the last minute?
    4 replies
    Hi everyone!
    Recently I've been testing kafka consumer reconnection after stopping and starting again kafka container. I'm using commitableExternalSource as a consumer.
    Use case:
    1. Start kafka in docker container.
    2. Start kafka consumer and producer (everything is working fine, messages are consumed).
    3. Stop kafka container (consumer is sending "Kafka commit is to be retried" logs (caused by DisconnectException)).
      And then:
      4.1. Start kafka again in ~10 sec after stopping (everything goes back to normal, consumer is working properly)
      4.2. Wait a bit longer with starting kafka containter:
      4.2.1. kafka is sending Compliting logs from BaseSingleSourceLogic class,
      4.2.2. "Kafka commit is to be retried" logs are still being sent,
      4.2.3. after starting kafka container again there are logs from deadLetter and no more logs are shown.
    Roy Prager
    i am consuming from a topic with multiple partitions, and using groupedWithin in my graph, means that i should commit the offset to multiple partitions. what is the best way for doing it?
    Hi guys!
    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