fs2-kafka/Lobby

Andi Miller

@andimiller

have you considered adding the timestamp field in the producer?

KSQL gets a bit upset if I read data written by fs2-kafka

Adam Chlupacek

@AdamChlupacek

We are adhering to the kafka protocol, if you need some timestamps in the messages written to kafka, you can just add it there yourself. Or am I missing something? I dont want to add some additional data to everything just because of KSQL.

Andi Miller

@andimiller

there is part of the kafka protocol which has timestamps I believe

but it's optional

https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message

Adam Chlupacek

@AdamChlupacek

Oooh, I see what you mean, we are just not populating the value, but it is in protocols. all that is required is to expose it via API as of now we jsut set it to None all the me, you can PR it if you want.

Andi Miller

@andimiller

I may do if I can make it work :)

Adam Chlupacek

@AdamChlupacek

Hmm actually this may required more thought to it, now that i am reading the link :/

Andi Miller

@andimiller

I think it's a case of adding a new codec in protocols

because it bumps the magic byte

Pavel Chlupacek

@pchlupacek

@andimiller any idea what version of kafka introduced this ?

Adam Chlupacek

@AdamChlupacek

In that case yeah, a whole new codec is required

Pavel Chlupacek

@pchlupacek

if you can please, could you open issue? We’ll look onto that

Andi Miller

@andimiller

in theory 0.10.2.0

Pavel Chlupacek

@pchlupacek

:-)

Andi Miller

@andimiller

will do, I'll try and gather some info

Pavel Chlupacek

@pchlupacek

excellent thanks a lot

Kåre Blakstad

@kareblak

Can you guys provide an example of fs2-kafka writing offsets back to kafka?

pacman899

@pacman899

HI, is there a way consuming one optional msg(latest) from a topic? maybe with timeout? it suppose to return Stream of one element of Option[Msg]

Pavel Chlupacek

@pchlupacek

@/all 0.4.0-M2 release agains fs2 1.0.0-M2 was published

Vinayak Pathak

@vinayakpathak

Hello. I'm trying to use version 0.2.0 and it looks like even getting kafka.client working requires a non-trivial amount of work. I seem to need the following implicits just to make it compile:

  implicit val _concurrent: Concurrent[IO] = IO.ioConcurrentEffect
  implicit val AG: AsynchronousChannelGroup = AsynchronousChannelGroup.withThreadPool(Executors.newFixedThreadPool(8))
  implicit val ec = ExecutionContext.fromExecutor(new ForkJoinPool(8))
  implicit val scheduler = Scheduler(corePoolSize = 8).compile.toList.unsafeRunSync().head
  implicit val logger = Logger.JDKLogger[IO](java.util.logging.Logger.getAnonymousLogger).unsafeRunSync()

I'm not really familiar with various concurrency and execution related functionalities in java and so I have no idea what I'm doing here. At the same time it seems to me that if these implicits were actually required it wouldve been mentioned on the readme? Am I doing something wrong?

Pavel Chlupacek

@pchlupacek

@vinayakpathak some of this is getting to be simplified with 0.4.0 release (also please do not use 0.2.x, but rather 0.4.x branch). However we prefer the configuration of concurrency to be explicit in most cases.

Vinayak Pathak

@vinayakpathak

I see. 0.4.x uses fs2 1.0.0, no? Is that considered stable yet?

Pavel Chlupacek

@pchlupacek

the last version of 1.0 has fixes that are not available in 0.10 and liekly won't be backported. The 1.0 still may have some API changes. We are using 1.0 in production atm.

Vinayak Pathak

@vinayakpathak

Okay I'll give it a shot then

Pavel Chlupacek

@pchlupacek

generally the 0.10 version of kafka was imho never used a lot. In our production code we were at 0.9 release for a long time, and now we went straight to 1.0

Vinayak Pathak

@vinayakpathak

you mean 0.10 version of fs2 right?

Pavel Chlupacek

@pchlupacek

yes

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

Vinayak Pathak

@vinayakpathak

fs2-kafka's readme says M4

looks like it's going to be fs2-kafka v0.4.0-M3, which will depend on fs2 1.0.0-M4

Pavel Chlupacek

@pchlupacek

This seems to be messed up. @AdamChlupacek can you please take a look?

Adam Chlupacek

@AdamChlupacek

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 :/

Pavel Chlupacek

@pchlupacek

I think m2 and m4 us not binary compatible.

Daniel Karch

@danielkarch

Hi, do you think a version compatible with fs2 1.0.0-M5 will be released soon?

Pavel Chlupacek

@pchlupacek

Sorry @danielkarch. At this moment we are trying to get last changes to fs2 to release RC1 of fs2. If possible that that would be the next target release for fs2-kafka. Sorry for inconvinience.

Daniel Karch

@danielkarch

Okay, thank you.

Calvin Brown

@calvinbrown085

Hi All, why does the publish* methods need the key and the value to be a ByteVector ?

Adam Chlupacek

@AdamChlupacek

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.

Calvin Brown

@calvinbrown085

:thumbsup:

Ernesto Periñán

@eperinan

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?

Pavel Chlupacek

@pchlupacek

@eperinan we do not implement kafka tooling in the fs2-kafka. So you have to create this manually. There is some tooling in tests that can help you do so, or you may use standard tooling available in kafka.

re the method, which one are you referring to ?

Ernesto Periñán

@eperinan

Thanks for your quick reply

This method https://github.com/Spinoco/fs2-kafka/blob/series/0.4/src/main/scala/spinoco/fs2/kafka/KafkaClient.scala#L110

I mean that we are not handling the exceptions in this method. IMHO, it is really intresting. What do you think?

Pavel Chlupacek

@pchlupacek

note that KafkaClient to be created requires an F : Concurrent, concurrent actually extends MonadError/ApplicativeError where E is defined as Throwable. So in fact this Either[Throwable, Long] is already there :-)

Where communities thrive

People

Repo info

Activity