kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Magnus Vojbacke <magnus.vojba...@digitalroute.com>
Subject Re: getOffsetsBefore(...) => kafka.common.UnknownException
Date Fri, 17 Oct 2014 11:44:31 GMT
Yes, I used TopicMetaDataRequest to determine the lead broker for the partition.

I’ve also compared the broker selected by my code with the output from "bin/kafka-topics.sh
--describe[...]" and as far as I can tell, it’s using the broker which is described as leader
for the topic.

/Magnus


On 17 Oct 2014, at 05:55, Jun Rao <junrao@gmail.com> wrote:

> The OffsetRequest can only be answered by the leader of the partition. Did
> you connect the SimpleConsumer to the leader broker? If not, you need to
> use TopicMetadataRequest to find out the leader broker first.
> 
> Thanks,
> 
> Jun
> 
> On Thu, Oct 16, 2014 at 3:56 AM, Magnus Vojbacke <
> magnus.vojbacke@digitalroute.com> wrote:
> 
>> Hi,
>> 
>> I’m trying to make a request for offset information from my broker, and I
>> get a kafka.common.UnknownException as the result.
>> 
>> I’m trying to use the Simple Consumer API
>> 
>> 
>> 
>>        val topicAndPartition = new TopicAndPartition(“topic3”, 0)
>>        val requestInfo = new java.util.HashMap[TopicAndPartition,
>> PartitionOffsetRequestInfo]()
>>        requestInfo.put(topicAndPartition, new
>> PartitionOffsetRequestInfo(kafka.api.OffsetRequest.EarliestTime, 1))
>> 
>>        val request = new kafka.javaapi.OffsetRequest(requestInfo,
>> kafka.api.OffsetRequest.CurrentVersion, clientName)
>> 
>>        import kafka.javaapi._
>>        // conn: kafka.javaapi.consumer.SimpleConsumer
>>        val response: OffsetResponse = conn.getOffsetsBefore(request)
>> 
>>        println("got response [" + response + “]")
>> 
>> 
>> 
>> Output:
>> got response [OffsetResponse(0,Map([test3,1] -> error:
>> kafka.common.UnknownException offsets: 0))]
>> 
>> 
>> I really can’t figure out why I’m getting this response. As far as I know,
>> “topic3” with partition “0” exists on the broker, and I can use
>> bin/kafka-console-consumer.sh to consume from it without any problems.
>> 
>> 
>> Is there any idea of what could cause this exception?
>> 
>> As it is right now, I’m not even sure if the request gets to the broker.
>> Is there any way of activating more verbose logs on the broker?
>> 
>> I think I’m using a trunk build (2.10-0.8.3-SNAPSHOT)
>> 
>> 
>> BR
>> /Magnus
>> 
>> 


Mime
View raw message