the 0.4.0 of fs2-kafka will get likely synced with latest fs2 next week,
so no you are i think on M3
1.0.0-M3
looks like it's going to be fs2-kafka v0.4.0-M3, which will depend on fs2 1.0.0-M4
Well kafka 0.4.0-M2 is compatible with fs2 1.0.0-M3, that is why there is not kafka M3. So either we skip M3 for kafka so that we sync the milestone tag with fs2, or just release M3 kafka to be compatible with M4 fs2. No idea if current kafka (M2) is compatible with fs2 M4 :/
Hey there @calvinbrown085 , There are two point to answer that, first of all, it is a
ByteVector, not a ByteBuffer of Array[Byte] as ByteVector provides us with much high level API to operate over the bytes in it. As well it is immutable, so it makes it much easier to reason about. Second, we do not take some abstract A, because we want the user to deal with encoding errors. This is basically the gist of it.
hi all!! I'm trying to publish to a topic that it's not created yet, expecting the producer be able to create on the fly. Does anyone addressing this problem before? I've been checking the method and it's returning and F[Long] but I'm consider IMHO it should return a F[Either[E,Long]]. In my case, I'm getting the error LeaderNotAvailable which it should be encapsulated within E. Am I missing something?? I know that README talks about "leaves it up to the user to choose any the retry mechanism". any suggestions?
re the method, which one are you referring to ?
So if you want to handle error just use
attempt and you get that either you asking for :-)
Really thanks @pchlupacek I didn't take the
concurrent into account
I am going to review it. Thanks, you are awesome :muscle:
once that will be resolved we will issue a release.
Sorry for that
Hey everybody, just looking to use fs2-kafka in a project and wondering about the ovotech/fs2-kafka project. It seems to wrap the kafka client drivers. Why would I use one over the other? https://github.com/ovotech/fs2-kafka
This seems to work:
Stream.eval(client.offsetRangeFor(topic("offsets"), partition(0))).flatMap { case (first, next) =>
val num = next - first
println(s"F: $first, L: $next, N: $num")
client.subscribe(topic("offsets"), partition(0), first).take(num)
}
