Also, I thought a bit more about the org.apache thing. This is clearly an
aesthetic thing, but I do feel aesthetics are important.
The arguments for the kafka.* package is basically that putting your domain
name in the namespace is a silly cargo cult thing that Java people do. I
have never seen a
Fine with most of the changes. Just a few questions/comments.
1. Would it be better to change Callback to ProducerCallback to distinguish
it from controller callback and potential future other types of callbacks
(e.g. consumer)?
2. If no key is present AND no partition id is present, partitions
(1) if the
replication factor is 1 and there is a broker failure, we can still route
the message, (2) if a bunch of producers are started at the same time, this
prevents them from picking up the same partition in a synchronized way.
You raised a good point, Jun. Regarding #1, shouldn't round
1. Yes, I think the name could be improved. However that name doesn't
really express what it does. What about RecordSentCallback?
2. As Neha said the nodes are initialized to a random position. Round robin
is preferable to random (lower variance, cheaper to compute, etc). Your
point about skipping
For (3) we could also do the following:
- On any retryable producer error, force a metadata refresh (in
handleProducerResponse).
- In handleMetadataResponse, the producer can (internally) close out
connections that are no longer valid. (i.e., connections to {old set
of leader brokers} - {new
Hey Joel,
I actually went with FutureRecordMetadata instead of
FutureRecordPosition but yes, as you say that object represents the
topic, partition, and offset and would be the place we would put any future
info we need to return to the user.
The problem with implementing cancel is that the
Hey Neha,
Basically partitionsForTopic will invoke
Metadata.fetch(topic).partitionsFor. So it will block on the first request
for a given topic while metadata is loaded. Each subsequent request will
return whatever metadata is present, the logic for refreshing metadata will
remain exactly as it
It *is* expected that the user call this method on
every single call and because we no longer control the partitioner
interface there will be no way to control this.
Make sense. This will ensure that new partitions are detected as defined by
topic.metadata.refresh.interval.ms.
There are a quite
About serialization, I am wondering if most people would try to have a
customized partioning mainly due to logic motivations, such that they want
some messages to go to the same partition; if that is true, they do not
really care about the actually partition id but all they would do is
specify the
Okay I posted a patch against trunk that carries out the refactoring
described above:
https://issues.apache.org/jira/browse/KAFKA-1227
Updated javadoc is here:
http://empathybox.com/kafka-javadoc
This touches a fair number of files as I also improved documentation and
standardized terminology in
.
In order to allow correctly choosing a partition the Producer interface
will include a new method:
ListPartitionInfo partitionsForTopic(String topic);
PartitionInfo will be changed to include the actual Node objects not just
the Node ids.
Why are the node id's alone insufficient?
1. Change send to use java.util.concurrent.Future in send():
FutureRecordPosition send(ProducerRecord record, Callback callback)
The cancel method will always return false and not do anything. Callback
will then be changed to
interface Callback {
public void onCompletion(RecordPosition
Hey folks,
Thanks for all the excellent suggestions on the producer API, I think this
really made things better. We'll do a similar thing for the consumer as we
get a proposal together. I wanted to summarize everything I heard and the
proposed changes I plan to do versus ignore. I'd like to get
13 matches
Mail list logo
