spinoco / fs2-kafka Goto Github PK

for example, in "publishN" function, i must mention a specific partition id.

is there a way to get around it ?

def publishN(
topicId: String @@ TopicName
, partition: Int @@ PartitionId
, requireQuorum: Boolean
, serverAckTimeout: FiniteDuration
, compress: Option[Compression.Value]
)(messages: Chunk[(ByteVector, ByteVector)]): F[Long]

Hi. I am getting very strange behaviour. Suppose we have one broker and we have subscribed to a topic there. We are happily consuming from this topic and then the broker goes down. After some time it is going up again and Kafka client is able to reconnect back, but it won't get any messages anymore, and there is no notification about this event that is exposed to fs2 code, so we can, for example, resubscribe to this topic, restart the consumer/stream or something like this.

I think this problem is common and someone can point me where to look for a solution (and maybe we can extend the documentation to cover this issue).

Hi
I am trying to use fs2-kafka as following:

import fs2._
import spinoco.fs2.kafka
import spinoco.fs2.kafka._
import spinoco.protocol.kafka._
import scala.concurrent.duration._
import scodec.bits.ByteVector


object Client {

  def main(args: Array[String]): Unit = {

    kafka.client(
      ensemble = Set(broker("localhost", port = 9092))
      , protocol = ProtocolVersion.Kafka_0_10_2
      , clientName = "Khinkali"
    ) flatMap { kafkaClient =>
      kafkaClient
        .publish1(topic("topic-A"), partition(0), ByteVector("foo"), ByteVector("Hello FOO"), requireQuorum = true, serverAckTimeout = 10 seconds)
        .asInstanceOf
    }
  }

}

But I've got following error:

[info] Compiling 1 Scala source to /home/developer/Desktop/scala/fskafka/target/scala-2.12/classes ...
[error] /home/developer/Desktop/scala/fskafka/src/main/scala/Client.scala:16:5: Symbol 'type fs2.util.Async' is missing from the classpath.
[error] This symbol is required by 'value spinoco.fs2.kafka.package.F'.
[error] Make sure that type Async is in your classpath and check for conflicting dependencies with `-Ylog-classpath`.
[error] A full rebuild may help if 'package.class' was compiled against an incompatible version of fs2.util.
[error]     kafka.client(
[error]     ^
[error] /home/developer/Desktop/scala/fskafka/src/main/scala/Client.scala:16:17: could not find implicit value for parameter AG: java.nio.channels.AsynchronousChannelGroup
[error]     kafka.client(
[error]                 ^
[error] two errors found
[error] (compile:compileIncremental) Compilation failed
[error] Total time: 1 s, completed Nov 13, 2017 12:17:13 PM
[IJ]sbt:fskafka> 

What am I doing wrong?
Thanks

When using offsetRangeFor the connections to brokers do not close up properly and are stuck in the "TIME_WAIT" state, this can be seen in netstat. The connections do die out after while.

There are two connections that get stuck like this. One is when we try to find a leader for the given topicId and second one is the actual fetch of the offsetRange.

There may be other instances where the connections do not close up properly. As such we should check every place where we communicate with kafka.

Hi - could you please tell me if this project is still under development?

I see the last commit was 7 months ago.

Thanks!

I'm still fairly new to this library, but Kafka released 1.0 this week. After doing #8 is there anything that needs to be done to support the 1.0 protocol? Is there anything to be gained? Should we at least test against it so we have something to say? Really I'm just looking for guidance here as a newcomer.

I found that after running a stream to get a nonempty collection of messages, the offsets weren't incremented. Is there a way in this client to commit offsets while the stream is running?

I used code something like the following, which threw NoOffsetForPartitionException. The groupIds on the clients are the same , and there was only one partition.

      val topicPartition = new TopicPartition(topic, 0)

      val receivedMessages = t.getKafkaMessages[IO]().take(messages.size).compile.toVector.unsafeRunSync()

      val consumerAfter =
        new KafkaConsumer(
          Map[String, AnyRef](
            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers,
            ConsumerConfig.RETRY_BACKOFF_MS_CONFIG -> "1000",
            ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> "false",
            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "none",
            ConsumerConfig.GROUP_ID_CONFIG -> "increments-offsets"
          ).asJava,
          new ByteArrayDeserializer(),
          Generic.kafkaDeserializer
        )
      consumerAfter.assign(List(topicPartition).asJava)
      val endOffset = consumerAfter.position(topicPartition)

Kafka added timestamps to it's messages back some time around 0.10.x

It's handy to have these so the server can compact more reliably, and so KSQL doesn't throw up when reading messages produced by fs2-kafka.

Useful links:
The original KIP
The JIRA ticket
The github PR with the change

Now that Cats 1.0.0-RC1 is out and fs2 0.10 is near final, it'd be nice to get a version built on top of those versions.

KafkaClient#subscribe scaladoc says the following:

-1 specified start from tail (new message arriving to topic)

This is wrong as this will start reading the topic from the beginning rather than tail according to code. The message should say something like following.

TailOffset specified start from tail (new message arriving to topic)

Hi
I wanted to ask, if it is save to use it in production?
Are you using in production?
Thanks

Hi. I was hoping to use the library against cloudkarafka but it uses an SSL connection. It looks like the library does not support that. Is there any plan to?

https://www.cloudkarafka.com/docs-java.html