fs2-kafka/Lobby

Andi Miller

@andimiller

it could be that I'm talking to a 1.0 server, but I'd expect that to work

Adam Chlupacek

@AdamChlupacek

we have tests on travis running agains 1.0, it works fine. could you show me where you are creating the kafka client?

Andi Miller

@andimiller

      import scala.concurrent.ExecutionContext.Implicits.global
      val pool = Executors.newScheduledThreadPool(8)
      implicit val scheduler = fs2.Scheduler.fromScheduledExecutorService(pool)
      implicit val ag = AsynchronousChannelGroup.withThreadPool(pool)
      kafka.client[IO](
        ensemble = Set(broker("localhost", port = 9092))
        , protocol = ProtocolVersion.Kafka_0_10_2
        , clientName = "testclient"
      )

do scheduler and ag have to be separate pools?

Adam Chlupacek

@AdamChlupacek

i would suggest that in the future, but that certainly does not cause this issue. Are you prehaps compiling and running the stream where the kafka client is created before you pass it to that send function?

Andi Miller

@andimiller

I am, yes

is it bracketed to shut it down?

Adam Chlupacek

@AdamChlupacek

yeah exactly. You cannot run the stream until the "end of the world", since kafka client is treated as a resource, thus it is being released when the stream ends.

Andi Miller

@andimiller

so if I can integrate it into my streamapp it should work?

Adam Chlupacek

@AdamChlupacek

yeah exactly

Andi Miller

@andimiller

cool, thanks, I'll give that a go :)

Adam Chlupacek

@AdamChlupacek

no problem, feel free to ask if you have any other questions :)

Andi Miller

@andimiller

any of you who're using fs2-kafka for consuming, what do you store your offsets in?

Adam Chlupacek

@AdamChlupacek

We use cassandra for storing any and all persistent data.

Andi Miller

@andimiller

so you commit offsets to cassandra when you've processed a batch?

the normal kafka client keeps it in kafka, somehow

and used to keep it in zookeeper

I guess cassandra makes sense since it's very write-heavy

Adam Chlupacek

@AdamChlupacek

basically yeah we keep it there, its needed for recovery etc, no idea how the client can keep anything like that when you crash mid batch or anything like that

Andi Miller

@andimiller

yeah of course

Adam Chlupacek

@AdamChlupacek

I may be misunderstanding here a bit tho, what for do you actually want to store the offset? (since you are comparing to the kafka client, you cannot be using it for recovery etc right? )

Andi Miller

@andimiller

I'm using the offset for crash recovery yeah

I'm currently using the official kafka consumer, and fs2-kafka's producer

so I don't have to implement my own offset management

Adam Chlupacek

@AdamChlupacek

What kind of crash recovery? complete (jvm exits)? or just an error in your stream?

Andi Miller

@andimiller

usually just a normal SIGTERM

the rpm being upgraded

Adam Chlupacek

@AdamChlupacek

And kafka client can somehow get up from this? wow, it must store it in zookeeper or something like that, seems a bit arcane to me :D

if you need something short term, you can store the offsets in zookeeper manually maybe? that would allow you to use fs2-kafka consumer as well.

Pavel Chlupacek

@pchlupacek

Actually you may use kafka to store offset, and also the fs2-kafka. Just configure one of your topics to be compressed and you are good to go. Thats actually how original client of kafka stores offset nowadays, before it was ZK.

Adam Chlupacek

@AdamChlupacek

didnt know that, thats cool.

Andi Miller

@andimiller

yeah I need to read up on how the official client stores it in kafka and do that

Pavel Chlupacek

@pchlupacek

for that you will need to read kafkas client source. Not sure if they document that.

fwiw if you will go this path I would be happy to have PR with that :-)

Andi Miller

@andimiller

I guess you key on consumerid, and compress it

Pavel Chlupacek

@pchlupacek

yes, I woudl say so

and perhaps topic

Filippo De Luca

@filosganga

Hi Guys,
Am I correct to say that to use fs2-kafka I need to assign a specific partition to each consumer?

Adam Chlupacek

@AdamChlupacek

yes, that is correct, you need to specify a partition.

Filippo De Luca

@filosganga

@AdamChlupacek Thanks

sken

@sken77

im thinking of doing a smack stack with just the fs2 libs

is this possible? or a good idea

Adam Chlupacek

@AdamChlupacek

Hey @floreslorca i am not entirely sure what you mean by smack, but with a quick google, it seesm that it is just a technique for distributed apps? In that case yeah you can. We in our production fs2 only. (As you can tell by all the fs2 libs we have on Spinoco here :D)

sken

@sken77

its so hot right now :p https://www.oreilly.com/ideas/the-smack-stack

i just despise akka and friends so i want to try it with fs2

sken

@sken77

the idea is very generic and reusable, and i was wondering if its doable with the fs2 libs

Adam Chlupacek

@AdamChlupacek

@floreslorca Yeah, all of this seems doable, you have cassandra and kafka covered with our libraries, akka with fs2, sadly I dont really have experience with mesos or spark, so cannot tell you how to replace that atm with some fs2 library. But from my experience you can express anything in fs2 with much better guarantees than any other lib :D And ofc in its very core fs2 is trying to be as generic an reusable as it can be.

sken

@sken77

i dont really need the spark part at the moment. mesos is just the environment is running on

Adam Chlupacek

@AdamChlupacek

In that case you can totally do this. Btw as far as I understand spark, its just some kind of concurrent data processing?

sken

@sken77

yep, thats basically what it is

Where communities thrive

People

Repo info

Activity