You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Gerrit Jansen van Vuuren <ge...@gmail.com> on 2014/10/12 23:22:25 UTC
kafka java api (written in 100% clojure)
Hi,
Just thought I'll put this out for the kafka community to see (if anyone
finds it useful great!!).
Kafka-fast is 100% pure clojure implementation for kafka, but not just
meant for clojure because it has a Java API wrapper that can be used from
Java, Groovy, JRuby or Scala.
This library does not wrap scala instead it directly communicates with the
kafka cluster.
It also uses redis for offsets instead of zookeeper and removes the one
consumer per partition limit that all other kafka libraries have, by diving
offsets into logical work units and running consumption via a queue(list)
in redis.
https://github.com/gerritjvv/kafka-fast
Any constructive critique is more than welcome :)
Re: kafka java api (written in 100% clojure)
Posted by Gerrit Jansen van Vuuren <ge...@gmail.com>.
Hi Gwen,
For other reasons :), I've used zookeeper (from Java) in the past to store
similar information and have found it doesnt work good enough for this use
case in particular. Zk is great for config and locks but not for data that
changes fast and can grow beyond a certain limit. In particular my worst
experience with zk has been sync data becoming too big to flush to disks
and snapshots taking too long if ever to be read by quorum nodes after a
leader election, sometimes the only real option was to delete all data from
disk and start a new fresh zookeeper cluster.
I've also moved away from the consumer = topic/partition constraint by
using a list/queue in redis (agreeably something I can also do in zk) but
redis again just feels like a better fit for it, especially with things
like brpoppush.
On Tue, Oct 14, 2014 at 12:09 AM, Gwen Shapira <gs...@cloudera.com>
wrote:
> Out of curiosity: did you choose Redis because ZooKeeper is not well
> supported in Clojure? Or were there other reasons?
>
> On Mon, Oct 13, 2014 at 2:04 PM, Gerrit Jansen van Vuuren
> <ge...@gmail.com> wrote:
> > Hi Steven,
> >
> > Redis:
> >
> > I've had a discussion on redis today, and one architecture that does
> come
> > up is using a master slave, then if the master fails the have the
> > application start writing to the slave. Writing to a slave is possible in
> > redis, albeit you cannot fail back to the master because writes to a
> slave
> > will not be automatically replicated to the master.
> >
> > Any suggestions are welcome.
> >
> > Java:
> >
> > I like Java, have been with it a long time, did groovy, went back to
> > Java, tried scala, went back to Java, then tried clojure and got sold on
> it
> > (mainly because it fits my way of thinking). OtherJVMLang -> Java interop
> > is always better than Java -> OtherJVMLang interop. Clojure interop to
> Java
> > is really great, but Java to Clojure you need to use things like:
> > RT.var("kafka-clj.consumer.node", "read-msg!").invoke(connector,
> > timeoutMillis))
> >
> >
> > I've refactored the Java API to be more "Java like" (plus made Consumer
> > Iterable), and made a new release "2.3.9", see the updated examples, also
> > have a look at
> >
> https://github.com/gerritjvv/kafka-fast/blob/master/kafka-clj/doc/vagrant.md
> > .
> >
> > My idea for this library is that from Java/Groovy etc you do not need to
> > know about Clojure behind the scenes (barring the stacktraces), you just
> > get Java, and obviously if your using Clojure you just get Clojure ;).
> >
> > Cheers,
> > Gerrit
> >
> >
> > On Mon, Oct 13, 2014 at 6:52 PM, Steven Schlansker <
> > sschlansker@opentable.com> wrote:
> >
> >> Couple of mostly-uninformed comments inline,
> >>
> >>
> >> On Oct 13, 2014, at 2:00 AM, Gerrit Jansen van Vuuren <
> gerritjvv@gmail.com>
> >> wrote:
> >>
> >> > Hi Daniel,
> >> >
> >> > At the moment redis is a spof in the architecture, but you can setup
> >> > replication and I'm seriously looking into using redis cluster to
> >> eliminate
> >> > this.
> >> > Some docs that point to this are:
> >> > http://redis.io/topics/cluster-tutorial
> >> > http://redis.io/topics/sentinel
> >>
> >> There's some evidence that redis clusters are *not* good for managing
> state
> >> in the way that you are using it:
> >>
> >> http://aphyr.com/posts/283-call-me-maybe-redis
> >>
> >> > If you can’t tolerate data loss, Redis Sentinel (and by extension
> Redis
> >> Cluster) is not safe for use as:
> >> >
> >> > • A lock service
> >> > • A queue
> >> > • A database
> >>
> >>
> >> >>
> >> >>> On 13/10/2014, at 10:22 am, Gerrit Jansen van Vuuren <
> >> >> gerritjvv@gmail.com> wrote:
> >> >>>
> >> >>> Hi,
> >> >>>
> >> >>> Just thought I'll put this out for the kafka community to see (if
> >> anyone
> >> >>> finds it useful great!!).
> >> >>>
> >> >>> Kafka-fast is 100% pure clojure implementation for kafka, but not
> just
> >> >>> meant for clojure because it has a Java API wrapper that can be used
> >> from
> >> >>> Java, Groovy, JRuby or Scala.
> >>
> >> One thing that frustrates me with the Kafka library is that despite it
> >> claiming
> >> that the Scala code is interoperable with Java, it really isn't. You
> end
> >> up
> >> having to work around the Scala compiler 'magic' in increasingly bizarre
> >> ways,
> >> e.g. default arguments:
> >>
> >> kafka = new KafkaServer(createConfig(), KafkaServer.init$default$2());
> >>
> >> which is both magical and fragile. I don't know whether Clojure is the
> >> same way,
> >> just want to point out that if you don't take particular care of us old
> >> fart Java
> >> nuts, you'll lose us quickly :)
> >>
> >> Another example, from your docs:
> >>
> >> Object connector = Producer.createConnector(new
> BrokerConf("192.168.4.40",
> >> 9092));
> >> Producer.sendMsg(connector, "my-topic", "Hi".getBytes("UTF-8"));
> >>
> >> This is downright bizarre to me, I would instead expect:
> >>
> >> Producer connector = Producer.createConnector(...)
> >> connector.sendMsg("my-topic", bytes)
> >>
> >> which is IMO shorter, cleaner, and easier for testing (especially
> mocking).
> >>
> >>
> >> Hope some of my ravings are helpful,
> >> Steven
> >>
> >>
>
Re: kafka java api (written in 100% clojure)
Posted by Gwen Shapira <gs...@cloudera.com>.
Out of curiosity: did you choose Redis because ZooKeeper is not well
supported in Clojure? Or were there other reasons?
On Mon, Oct 13, 2014 at 2:04 PM, Gerrit Jansen van Vuuren
<ge...@gmail.com> wrote:
> Hi Steven,
>
> Redis:
>
> I've had a discussion on redis today, and one architecture that does come
> up is using a master slave, then if the master fails the have the
> application start writing to the slave. Writing to a slave is possible in
> redis, albeit you cannot fail back to the master because writes to a slave
> will not be automatically replicated to the master.
>
> Any suggestions are welcome.
>
> Java:
>
> I like Java, have been with it a long time, did groovy, went back to
> Java, tried scala, went back to Java, then tried clojure and got sold on it
> (mainly because it fits my way of thinking). OtherJVMLang -> Java interop
> is always better than Java -> OtherJVMLang interop. Clojure interop to Java
> is really great, but Java to Clojure you need to use things like:
> RT.var("kafka-clj.consumer.node", "read-msg!").invoke(connector,
> timeoutMillis))
>
>
> I've refactored the Java API to be more "Java like" (plus made Consumer
> Iterable), and made a new release "2.3.9", see the updated examples, also
> have a look at
> https://github.com/gerritjvv/kafka-fast/blob/master/kafka-clj/doc/vagrant.md
> .
>
> My idea for this library is that from Java/Groovy etc you do not need to
> know about Clojure behind the scenes (barring the stacktraces), you just
> get Java, and obviously if your using Clojure you just get Clojure ;).
>
> Cheers,
> Gerrit
>
>
> On Mon, Oct 13, 2014 at 6:52 PM, Steven Schlansker <
> sschlansker@opentable.com> wrote:
>
>> Couple of mostly-uninformed comments inline,
>>
>>
>> On Oct 13, 2014, at 2:00 AM, Gerrit Jansen van Vuuren <ge...@gmail.com>
>> wrote:
>>
>> > Hi Daniel,
>> >
>> > At the moment redis is a spof in the architecture, but you can setup
>> > replication and I'm seriously looking into using redis cluster to
>> eliminate
>> > this.
>> > Some docs that point to this are:
>> > http://redis.io/topics/cluster-tutorial
>> > http://redis.io/topics/sentinel
>>
>> There's some evidence that redis clusters are *not* good for managing state
>> in the way that you are using it:
>>
>> http://aphyr.com/posts/283-call-me-maybe-redis
>>
>> > If you can’t tolerate data loss, Redis Sentinel (and by extension Redis
>> Cluster) is not safe for use as:
>> >
>> > • A lock service
>> > • A queue
>> > • A database
>>
>>
>> >>
>> >>> On 13/10/2014, at 10:22 am, Gerrit Jansen van Vuuren <
>> >> gerritjvv@gmail.com> wrote:
>> >>>
>> >>> Hi,
>> >>>
>> >>> Just thought I'll put this out for the kafka community to see (if
>> anyone
>> >>> finds it useful great!!).
>> >>>
>> >>> Kafka-fast is 100% pure clojure implementation for kafka, but not just
>> >>> meant for clojure because it has a Java API wrapper that can be used
>> from
>> >>> Java, Groovy, JRuby or Scala.
>>
>> One thing that frustrates me with the Kafka library is that despite it
>> claiming
>> that the Scala code is interoperable with Java, it really isn't. You end
>> up
>> having to work around the Scala compiler 'magic' in increasingly bizarre
>> ways,
>> e.g. default arguments:
>>
>> kafka = new KafkaServer(createConfig(), KafkaServer.init$default$2());
>>
>> which is both magical and fragile. I don't know whether Clojure is the
>> same way,
>> just want to point out that if you don't take particular care of us old
>> fart Java
>> nuts, you'll lose us quickly :)
>>
>> Another example, from your docs:
>>
>> Object connector = Producer.createConnector(new BrokerConf("192.168.4.40",
>> 9092));
>> Producer.sendMsg(connector, "my-topic", "Hi".getBytes("UTF-8"));
>>
>> This is downright bizarre to me, I would instead expect:
>>
>> Producer connector = Producer.createConnector(...)
>> connector.sendMsg("my-topic", bytes)
>>
>> which is IMO shorter, cleaner, and easier for testing (especially mocking).
>>
>>
>> Hope some of my ravings are helpful,
>> Steven
>>
>>
Re: kafka java api (written in 100% clojure)
Posted by Gerrit Jansen van Vuuren <ge...@gmail.com>.
Hi Steven,
Redis:
I've had a discussion on redis today, and one architecture that does come
up is using a master slave, then if the master fails the have the
application start writing to the slave. Writing to a slave is possible in
redis, albeit you cannot fail back to the master because writes to a slave
will not be automatically replicated to the master.
Any suggestions are welcome.
Java:
I like Java, have been with it a long time, did groovy, went back to
Java, tried scala, went back to Java, then tried clojure and got sold on it
(mainly because it fits my way of thinking). OtherJVMLang -> Java interop
is always better than Java -> OtherJVMLang interop. Clojure interop to Java
is really great, but Java to Clojure you need to use things like:
RT.var("kafka-clj.consumer.node", "read-msg!").invoke(connector,
timeoutMillis))
I've refactored the Java API to be more "Java like" (plus made Consumer
Iterable), and made a new release "2.3.9", see the updated examples, also
have a look at
https://github.com/gerritjvv/kafka-fast/blob/master/kafka-clj/doc/vagrant.md
.
My idea for this library is that from Java/Groovy etc you do not need to
know about Clojure behind the scenes (barring the stacktraces), you just
get Java, and obviously if your using Clojure you just get Clojure ;).
Cheers,
Gerrit
On Mon, Oct 13, 2014 at 6:52 PM, Steven Schlansker <
sschlansker@opentable.com> wrote:
> Couple of mostly-uninformed comments inline,
>
>
> On Oct 13, 2014, at 2:00 AM, Gerrit Jansen van Vuuren <ge...@gmail.com>
> wrote:
>
> > Hi Daniel,
> >
> > At the moment redis is a spof in the architecture, but you can setup
> > replication and I'm seriously looking into using redis cluster to
> eliminate
> > this.
> > Some docs that point to this are:
> > http://redis.io/topics/cluster-tutorial
> > http://redis.io/topics/sentinel
>
> There's some evidence that redis clusters are *not* good for managing state
> in the way that you are using it:
>
> http://aphyr.com/posts/283-call-me-maybe-redis
>
> > If you can’t tolerate data loss, Redis Sentinel (and by extension Redis
> Cluster) is not safe for use as:
> >
> > • A lock service
> > • A queue
> > • A database
>
>
> >>
> >>> On 13/10/2014, at 10:22 am, Gerrit Jansen van Vuuren <
> >> gerritjvv@gmail.com> wrote:
> >>>
> >>> Hi,
> >>>
> >>> Just thought I'll put this out for the kafka community to see (if
> anyone
> >>> finds it useful great!!).
> >>>
> >>> Kafka-fast is 100% pure clojure implementation for kafka, but not just
> >>> meant for clojure because it has a Java API wrapper that can be used
> from
> >>> Java, Groovy, JRuby or Scala.
>
> One thing that frustrates me with the Kafka library is that despite it
> claiming
> that the Scala code is interoperable with Java, it really isn't. You end
> up
> having to work around the Scala compiler 'magic' in increasingly bizarre
> ways,
> e.g. default arguments:
>
> kafka = new KafkaServer(createConfig(), KafkaServer.init$default$2());
>
> which is both magical and fragile. I don't know whether Clojure is the
> same way,
> just want to point out that if you don't take particular care of us old
> fart Java
> nuts, you'll lose us quickly :)
>
> Another example, from your docs:
>
> Object connector = Producer.createConnector(new BrokerConf("192.168.4.40",
> 9092));
> Producer.sendMsg(connector, "my-topic", "Hi".getBytes("UTF-8"));
>
> This is downright bizarre to me, I would instead expect:
>
> Producer connector = Producer.createConnector(...)
> connector.sendMsg("my-topic", bytes)
>
> which is IMO shorter, cleaner, and easier for testing (especially mocking).
>
>
> Hope some of my ravings are helpful,
> Steven
>
>
Re: kafka java api (written in 100% clojure)
Posted by Steven Schlansker <ss...@opentable.com>.
Couple of mostly-uninformed comments inline,
On Oct 13, 2014, at 2:00 AM, Gerrit Jansen van Vuuren <ge...@gmail.com> wrote:
> Hi Daniel,
>
> At the moment redis is a spof in the architecture, but you can setup
> replication and I'm seriously looking into using redis cluster to eliminate
> this.
> Some docs that point to this are:
> http://redis.io/topics/cluster-tutorial
> http://redis.io/topics/sentinel
There's some evidence that redis clusters are *not* good for managing state
in the way that you are using it:
http://aphyr.com/posts/283-call-me-maybe-redis
> If you can’t tolerate data loss, Redis Sentinel (and by extension Redis Cluster) is not safe for use as:
>
> • A lock service
> • A queue
> • A database
>>
>>> On 13/10/2014, at 10:22 am, Gerrit Jansen van Vuuren <
>> gerritjvv@gmail.com> wrote:
>>>
>>> Hi,
>>>
>>> Just thought I'll put this out for the kafka community to see (if anyone
>>> finds it useful great!!).
>>>
>>> Kafka-fast is 100% pure clojure implementation for kafka, but not just
>>> meant for clojure because it has a Java API wrapper that can be used from
>>> Java, Groovy, JRuby or Scala.
One thing that frustrates me with the Kafka library is that despite it claiming
that the Scala code is interoperable with Java, it really isn't. You end up
having to work around the Scala compiler 'magic' in increasingly bizarre ways,
e.g. default arguments:
kafka = new KafkaServer(createConfig(), KafkaServer.init$default$2());
which is both magical and fragile. I don't know whether Clojure is the same way,
just want to point out that if you don't take particular care of us old fart Java
nuts, you'll lose us quickly :)
Another example, from your docs:
Object connector = Producer.createConnector(new BrokerConf("192.168.4.40", 9092));
Producer.sendMsg(connector, "my-topic", "Hi".getBytes("UTF-8"));
This is downright bizarre to me, I would instead expect:
Producer connector = Producer.createConnector(...)
connector.sendMsg("my-topic", bytes)
which is IMO shorter, cleaner, and easier for testing (especially mocking).
Hope some of my ravings are helpful,
Steven
Re: kafka java api (written in 100% clojure)
Posted by Gerrit Jansen van Vuuren <ge...@gmail.com>.
Hi Daniel,
At the moment redis is a spof in the architecture, but you can setup
replication and I'm seriously looking into using redis cluster to eliminate
this.
Some docs that point to this are:
http://redis.io/topics/cluster-tutorial
http://redis.io/topics/sentinel
Consumer:
Consumption is plit into logical work units by default 10K offsets.
If redis fails, the messages currently read will all be consumed, while the
redis connection threads go mad trying to reconnect.
No data loss should occur (albeit I'm still setting up scenarios in which I
will test this).
The consumer threads read work assignments from a redis list
using brpoplpush (see http://redis.io/commands/BRPOPLPUSH) and with this
pushes
the current working unit into a "working" queue, if any errors are
encountered while consuming the current work unit will not be marked done,
on recovery (startup)
these "working" queues are scanned and any work units found are placed back
onto the primary queue ready for consumption.
If an error message from kafka is received e.g error-code 6, the consumer
will try to recreate the metadata for a work-unit, while this is done the
work-unit will reside on
a "working" queue.
Another idea I'm going to play with is disabling consumption per
machine/topic to allow a machine to completely drain its messages in a
normal operational way,
so that safe shutdown can be scheduled for any machine without having to
rely on multiple threads synching etc.
Producer:
All sends are async and a broker is randomly selected.
The producer has a retry cache (implemented using http://www.mapdb.org/),
if any failure during sending (network or connection errors) the message
is saved and retried.
If acks==1 and a failure response from kafka is received, the message is
retried N times.
The low-level producer api allows you to by-pass all this logic and send to
any producer you want, the message send is still async but you can
wait/block on a response from the server
and then react as required.
Auditing:
There is an event channel (clojure.core.async) that receives work-units
consumed with a status field for errors.
This can be used to auditing consumers, I normally send this data to a
riemann instance, or just to disk.
Redis is fast and stable and I've found that even running it on a mixed
service box (e.g along side mysql etc) works fine, but this doesn't mean
that the physical hardware underneath can't fail.
In my case with 2 years of production use of redis I've only had one outage
(due to hardware failure).
Hope this answers some of your questions.
Any ideas of improving this is welcome and feel free to contribute and/or
create issue tickets with possible features ;)
Regards,
On Mon, Oct 13, 2014 at 9:58 AM, Daniel Compton <de...@danielcompton.net>
wrote:
> Hi Gerrit
>
> Thanks for your contribution, I'm sure everyone here appreciates it,
> especially Clojure developers like myself. I do have one question: what are
> the guarantees you offer to users of your library under failures,
> particularly when Redis fails?
>
> --
> Daniel
>
> > On 13/10/2014, at 10:22 am, Gerrit Jansen van Vuuren <
> gerritjvv@gmail.com> wrote:
> >
> > Hi,
> >
> > Just thought I'll put this out for the kafka community to see (if anyone
> > finds it useful great!!).
> >
> > Kafka-fast is 100% pure clojure implementation for kafka, but not just
> > meant for clojure because it has a Java API wrapper that can be used from
> > Java, Groovy, JRuby or Scala.
> >
> > This library does not wrap scala instead it directly communicates with
> the
> > kafka cluster.
> >
> > It also uses redis for offsets instead of zookeeper and removes the one
> > consumer per partition limit that all other kafka libraries have, by
> diving
> > offsets into logical work units and running consumption via a queue(list)
> > in redis.
> >
> > https://github.com/gerritjvv/kafka-fast
> >
> > Any constructive critique is more than welcome :)
>
Re: kafka java api (written in 100% clojure)
Posted by Daniel Compton <de...@danielcompton.net>.
Hi Gerrit
Thanks for your contribution, I'm sure everyone here appreciates it, especially Clojure developers like myself. I do have one question: what are the guarantees you offer to users of your library under failures, particularly when Redis fails?
--
Daniel
> On 13/10/2014, at 10:22 am, Gerrit Jansen van Vuuren <ge...@gmail.com> wrote:
>
> Hi,
>
> Just thought I'll put this out for the kafka community to see (if anyone
> finds it useful great!!).
>
> Kafka-fast is 100% pure clojure implementation for kafka, but not just
> meant for clojure because it has a Java API wrapper that can be used from
> Java, Groovy, JRuby or Scala.
>
> This library does not wrap scala instead it directly communicates with the
> kafka cluster.
>
> It also uses redis for offsets instead of zookeeper and removes the one
> consumer per partition limit that all other kafka libraries have, by diving
> offsets into logical work units and running consumption via a queue(list)
> in redis.
>
> https://github.com/gerritjvv/kafka-fast
>
> Any constructive critique is more than welcome :)