You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by chenlax <la...@hotmail.com> on 2014/08/22 16:36:19 UTC
read message use SimpleConsumer,can not start with assign offset
kafka version is 0.8.0, the code as https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
when i assign a offset ,it will throw java.lang.ArrayIndexOutOfBoundsException: 0
java.lang.ArrayIndexOutOfBoundsException: 0
at com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
at com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
at com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
##EarliestTime : -2 || -1 #### The Start Offset : -2
Error fetching data from the Broker:10.209.22.206 Reason: 1
how can is get the message which is assign offset?
Thanks，
Lax
RE: read message use SimpleConsumer,can not start with assign offset
Posted by chenlax <la...@hotmail.com>.
i just want get message start with assign offse.
in the code , requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
"whichTime" is a message's offset
Thanks，
Lax
> Date: Sun, 24 Aug 2014 21:41:47 -0700
> Subject: Re: read message use SimpleConsumer,can not start with assign offset
> From: junrao@gmail.com
> To: users@kafka.apache.org
>
> If you want to get the latest offset, whichTime should be set to -1.
>
> Thanks,
>
> Jun
>
>
> On Sun, Aug 24, 2014 at 8:16 PM, chenlax <la...@hotmail.com> wrote:
>
> > yes, it's "return offsets[0]".whichTime is a long num, is message offset i
> > cached,and the kafka keep message 7days.
> >
> > Thanks，
> > Lax
> >
> >
> > > Date: Sun, 24 Aug 2014 19:58:35 -0700
> > > Subject: Re: read message use SimpleConsumer,can not start with assign
> > offset
> > > From: junrao@gmail.com
> > > To: users@kafka.apache.org
> > >
> > > Is line 64 "return offsets[0]"? What value did you pass into whichTime?
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > >
> > > On Sun, Aug 24, 2014 at 7:24 PM, chenlax <la...@hotmail.com> wrote:
> > >
> > > > the method getLastOffset as follow:
> > > >
> > > > public static long getLastOffset(SimpleConsumer consumer, String topic,
> > > > int partition,long whichTime, String clientName)
> > > > {
> > > > long l_ret = kafka.api.OffsetRequest.EarliestTime();
> > > > System.out.println("whichTime = " + whichTime);
> > > > TopicAndPartition topicAndPartition = new
> > TopicAndPartition(topic,
> > > > partition);
> > > > Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo
> > =
> > > > new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
> > > >
> > > > requestInfo.put(topicAndPartition, new
> > > > PartitionOffsetRequestInfo(whichTime, 1));
> > > > kafka.javaapi.OffsetRequest request = new
> > > > kafka.javaapi.OffsetRequest(requestInfo,
> > > > kafka.api.OffsetRequest.CurrentVersion(),clientName);
> > > > OffsetResponse response = consumer.getOffsetsBefore(request);
> > > > if (response.hasError()) {
> > > > System.out.println("Error fetching data Offset Data the
> > > > Broker. Reason: " + response.errorCode(topic, partition) + " | " +
> > > > response.toString() );
> > > > return 0;
> > > > }
> > > > long[] offsets = response.offsets(topic, partition);
> > > > return offsets[0];
> > > > }
> > > >
> > > > Thanks，
> > > > Lax
> > > >
> > > >
> > > > > Date: Fri, 22 Aug 2014 21:27:58 -0700
> > > > > Subject: Re: read message use SimpleConsumer,can not start with
> > assign
> > > > offset
> > > > > From: junrao@gmail.com
> > > > > To: users@kafka.apache.org
> > > > >
> > > > > What's code in line 164 of SimpleConsumerClient.getLastOffset()?
> > > > >
> > > > > Thanks,
> > > > >
> > > > > Jun
> > > > >
> > > > >
> > > > > On Fri, Aug 22, 2014 at 7:36 AM, chenlax <la...@hotmail.com> wrote:
> > > > >
> > > > > > kafka version is 0.8.0, the code as
> > > > > >
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> > > > > >
> > > > > > when i assign a offset ,it will throw
> > > > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > > > >
> > > > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > > > > at
> > > > > >
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
> > > > > > at
> > > > > >
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
> > > > > > at
> > > > > >
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
> > > > > > ##EarliestTime : -2 || -1 #### The Start Offset : -2
> > > > > > Error fetching data from the Broker:10.209.22.206 Reason: 1
> > > > > >
> > > > > > how can is get the message which is assign offset?
> > > > > >
> > > > > > Thanks，
> > > > > > Lax
> > > > > >
> > > >
> > > >
> >
> >
RE: read message use SimpleConsumer,can not start with assign offset
Posted by chenlax <la...@hotmail.com>.
thanks junrao.
it can't use assign offset to get the last offset,my mistake.
Thanks，
Lax
> Date: Sun, 24 Aug 2014 21:41:47 -0700
> Subject: Re: read message use SimpleConsumer,can not start with assign offset
> From: junrao@gmail.com
> To: users@kafka.apache.org
>
> If you want to get the latest offset, whichTime should be set to -1.
>
> Thanks,
>
> Jun
>
>
> On Sun, Aug 24, 2014 at 8:16 PM, chenlax <la...@hotmail.com> wrote:
>
> > yes, it's "return offsets[0]".whichTime is a long num, is message offset i
> > cached,and the kafka keep message 7days.
> >
> > Thanks，
> > Lax
> >
> >
> > > Date: Sun, 24 Aug 2014 19:58:35 -0700
> > > Subject: Re: read message use SimpleConsumer,can not start with assign
> > offset
> > > From: junrao@gmail.com
> > > To: users@kafka.apache.org
> > >
> > > Is line 64 "return offsets[0]"? What value did you pass into whichTime?
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > >
> > > On Sun, Aug 24, 2014 at 7:24 PM, chenlax <la...@hotmail.com> wrote:
> > >
> > > > the method getLastOffset as follow:
> > > >
> > > > public static long getLastOffset(SimpleConsumer consumer, String topic,
> > > > int partition,long whichTime, String clientName)
> > > > {
> > > > long l_ret = kafka.api.OffsetRequest.EarliestTime();
> > > > System.out.println("whichTime = " + whichTime);
> > > > TopicAndPartition topicAndPartition = new
> > TopicAndPartition(topic,
> > > > partition);
> > > > Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo
> > =
> > > > new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
> > > >
> > > > requestInfo.put(topicAndPartition, new
> > > > PartitionOffsetRequestInfo(whichTime, 1));
> > > > kafka.javaapi.OffsetRequest request = new
> > > > kafka.javaapi.OffsetRequest(requestInfo,
> > > > kafka.api.OffsetRequest.CurrentVersion(),clientName);
> > > > OffsetResponse response = consumer.getOffsetsBefore(request);
> > > > if (response.hasError()) {
> > > > System.out.println("Error fetching data Offset Data the
> > > > Broker. Reason: " + response.errorCode(topic, partition) + " | " +
> > > > response.toString() );
> > > > return 0;
> > > > }
> > > > long[] offsets = response.offsets(topic, partition);
> > > > return offsets[0];
> > > > }
> > > >
> > > > Thanks，
> > > > Lax
> > > >
> > > >
> > > > > Date: Fri, 22 Aug 2014 21:27:58 -0700
> > > > > Subject: Re: read message use SimpleConsumer,can not start with
> > assign
> > > > offset
> > > > > From: junrao@gmail.com
> > > > > To: users@kafka.apache.org
> > > > >
> > > > > What's code in line 164 of SimpleConsumerClient.getLastOffset()?
> > > > >
> > > > > Thanks,
> > > > >
> > > > > Jun
> > > > >
> > > > >
> > > > > On Fri, Aug 22, 2014 at 7:36 AM, chenlax <la...@hotmail.com> wrote:
> > > > >
> > > > > > kafka version is 0.8.0, the code as
> > > > > >
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> > > > > >
> > > > > > when i assign a offset ,it will throw
> > > > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > > > >
> > > > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > > > > at
> > > > > >
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
> > > > > > at
> > > > > >
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
> > > > > > at
> > > > > >
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
> > > > > > ##EarliestTime : -2 || -1 #### The Start Offset : -2
> > > > > > Error fetching data from the Broker:10.209.22.206 Reason: 1
> > > > > >
> > > > > > how can is get the message which is assign offset?
> > > > > >
> > > > > > Thanks，
> > > > > > Lax
> > > > > >
> > > >
> > > >
> >
> >
Re: read message use SimpleConsumer,can not start with assign offset
Posted by Jun Rao <ju...@gmail.com>.
If you want to get the latest offset, whichTime should be set to -1.
Thanks,
Jun
On Sun, Aug 24, 2014 at 8:16 PM, chenlax <la...@hotmail.com> wrote:
> yes, it's "return offsets[0]".whichTime is a long num, is message offset i
> cached,and the kafka keep message 7days.
>
> Thanks，
> Lax
>
>
> > Date: Sun, 24 Aug 2014 19:58:35 -0700
> > Subject: Re: read message use SimpleConsumer,can not start with assign
> offset
> > From: junrao@gmail.com
> > To: users@kafka.apache.org
> >
> > Is line 64 "return offsets[0]"? What value did you pass into whichTime?
> >
> > Thanks,
> >
> > Jun
> >
> >
> > On Sun, Aug 24, 2014 at 7:24 PM, chenlax <la...@hotmail.com> wrote:
> >
> > > the method getLastOffset as follow:
> > >
> > > public static long getLastOffset(SimpleConsumer consumer, String topic,
> > > int partition,long whichTime, String clientName)
> > > {
> > > long l_ret = kafka.api.OffsetRequest.EarliestTime();
> > > System.out.println("whichTime = " + whichTime);
> > > TopicAndPartition topicAndPartition = new
> TopicAndPartition(topic,
> > > partition);
> > > Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo
> =
> > > new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
> > >
> > > requestInfo.put(topicAndPartition, new
> > > PartitionOffsetRequestInfo(whichTime, 1));
> > > kafka.javaapi.OffsetRequest request = new
> > > kafka.javaapi.OffsetRequest(requestInfo,
> > > kafka.api.OffsetRequest.CurrentVersion(),clientName);
> > > OffsetResponse response = consumer.getOffsetsBefore(request);
> > > if (response.hasError()) {
> > > System.out.println("Error fetching data Offset Data the
> > > Broker. Reason: " + response.errorCode(topic, partition) + " | " +
> > > response.toString() );
> > > return 0;
> > > }
> > > long[] offsets = response.offsets(topic, partition);
> > > return offsets[0];
> > > }
> > >
> > > Thanks，
> > > Lax
> > >
> > >
> > > > Date: Fri, 22 Aug 2014 21:27:58 -0700
> > > > Subject: Re: read message use SimpleConsumer,can not start with
> assign
> > > offset
> > > > From: junrao@gmail.com
> > > > To: users@kafka.apache.org
> > > >
> > > > What's code in line 164 of SimpleConsumerClient.getLastOffset()?
> > > >
> > > > Thanks,
> > > >
> > > > Jun
> > > >
> > > >
> > > > On Fri, Aug 22, 2014 at 7:36 AM, chenlax <la...@hotmail.com> wrote:
> > > >
> > > > > kafka version is 0.8.0, the code as
> > > > >
> > >
> https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> > > > >
> > > > > when i assign a offset ,it will throw
> > > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > > >
> > > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > > > at
> > > > >
> > >
> com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
> > > > > at
> > > > >
> > >
> com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
> > > > > at
> > > > >
> > >
> com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
> > > > > ##EarliestTime : -2 || -1 #### The Start Offset : -2
> > > > > Error fetching data from the Broker:10.209.22.206 Reason: 1
> > > > >
> > > > > how can is get the message which is assign offset?
> > > > >
> > > > > Thanks，
> > > > > Lax
> > > > >
> > >
> > >
>
>
RE: read message use SimpleConsumer,can not start with assign offset
Posted by chenlax <la...@hotmail.com>.
yes, it's "return offsets[0]".whichTime is a long num, is message offset i cached,and the kafka keep message 7days.
Thanks，
Lax
> Date: Sun, 24 Aug 2014 19:58:35 -0700
> Subject: Re: read message use SimpleConsumer,can not start with assign offset
> From: junrao@gmail.com
> To: users@kafka.apache.org
>
> Is line 64 "return offsets[0]"? What value did you pass into whichTime?
>
> Thanks,
>
> Jun
>
>
> On Sun, Aug 24, 2014 at 7:24 PM, chenlax <la...@hotmail.com> wrote:
>
> > the method getLastOffset as follow:
> >
> > public static long getLastOffset(SimpleConsumer consumer, String topic,
> > int partition,long whichTime, String clientName)
> > {
> > long l_ret = kafka.api.OffsetRequest.EarliestTime();
> > System.out.println("whichTime = " + whichTime);
> > TopicAndPartition topicAndPartition = new TopicAndPartition(topic,
> > partition);
> > Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo =
> > new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
> >
> > requestInfo.put(topicAndPartition, new
> > PartitionOffsetRequestInfo(whichTime, 1));
> > kafka.javaapi.OffsetRequest request = new
> > kafka.javaapi.OffsetRequest(requestInfo,
> > kafka.api.OffsetRequest.CurrentVersion(),clientName);
> > OffsetResponse response = consumer.getOffsetsBefore(request);
> > if (response.hasError()) {
> > System.out.println("Error fetching data Offset Data the
> > Broker. Reason: " + response.errorCode(topic, partition) + " | " +
> > response.toString() );
> > return 0;
> > }
> > long[] offsets = response.offsets(topic, partition);
> > return offsets[0];
> > }
> >
> > Thanks，
> > Lax
> >
> >
> > > Date: Fri, 22 Aug 2014 21:27:58 -0700
> > > Subject: Re: read message use SimpleConsumer,can not start with assign
> > offset
> > > From: junrao@gmail.com
> > > To: users@kafka.apache.org
> > >
> > > What's code in line 164 of SimpleConsumerClient.getLastOffset()?
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > >
> > > On Fri, Aug 22, 2014 at 7:36 AM, chenlax <la...@hotmail.com> wrote:
> > >
> > > > kafka version is 0.8.0, the code as
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> > > >
> > > > when i assign a offset ,it will throw
> > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > >
> > > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > > at
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
> > > > at
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
> > > > at
> > > >
> > com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
> > > > ##EarliestTime : -2 || -1 #### The Start Offset : -2
> > > > Error fetching data from the Broker:10.209.22.206 Reason: 1
> > > >
> > > > how can is get the message which is assign offset?
> > > >
> > > > Thanks，
> > > > Lax
> > > >
> >
> >
Re: read message use SimpleConsumer,can not start with assign offset
Posted by Jun Rao <ju...@gmail.com>.
Is line 64 "return offsets[0]"? What value did you pass into whichTime?
Thanks,
Jun
On Sun, Aug 24, 2014 at 7:24 PM, chenlax <la...@hotmail.com> wrote:
> the method getLastOffset as follow:
>
> public static long getLastOffset(SimpleConsumer consumer, String topic,
> int partition,long whichTime, String clientName)
> {
> long l_ret = kafka.api.OffsetRequest.EarliestTime();
> System.out.println("whichTime = " + whichTime);
> TopicAndPartition topicAndPartition = new TopicAndPartition(topic,
> partition);
> Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo =
> new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
>
> requestInfo.put(topicAndPartition, new
> PartitionOffsetRequestInfo(whichTime, 1));
> kafka.javaapi.OffsetRequest request = new
> kafka.javaapi.OffsetRequest(requestInfo,
> kafka.api.OffsetRequest.CurrentVersion(),clientName);
> OffsetResponse response = consumer.getOffsetsBefore(request);
> if (response.hasError()) {
> System.out.println("Error fetching data Offset Data the
> Broker. Reason: " + response.errorCode(topic, partition) + " | " +
> response.toString() );
> return 0;
> }
> long[] offsets = response.offsets(topic, partition);
> return offsets[0];
> }
>
> Thanks，
> Lax
>
>
> > Date: Fri, 22 Aug 2014 21:27:58 -0700
> > Subject: Re: read message use SimpleConsumer,can not start with assign
> offset
> > From: junrao@gmail.com
> > To: users@kafka.apache.org
> >
> > What's code in line 164 of SimpleConsumerClient.getLastOffset()?
> >
> > Thanks,
> >
> > Jun
> >
> >
> > On Fri, Aug 22, 2014 at 7:36 AM, chenlax <la...@hotmail.com> wrote:
> >
> > > kafka version is 0.8.0, the code as
> > >
> https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> > >
> > > when i assign a offset ,it will throw
> > > java.lang.ArrayIndexOutOfBoundsException: 0
> > >
> > > java.lang.ArrayIndexOutOfBoundsException: 0
> > > at
> > >
> com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
> > > at
> > >
> com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
> > > at
> > >
> com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
> > > ##EarliestTime : -2 || -1 #### The Start Offset : -2
> > > Error fetching data from the Broker:10.209.22.206 Reason: 1
> > >
> > > how can is get the message which is assign offset?
> > >
> > > Thanks，
> > > Lax
> > >
>
>
RE: read message use SimpleConsumer,can not start with assign offset
Posted by chenlax <la...@hotmail.com>.
the method getLastOffset as follow:
public static long getLastOffset(SimpleConsumer consumer, String topic, int partition,long whichTime, String clientName)
{
long l_ret = kafka.api.OffsetRequest.EarliestTime();
System.out.println("whichTime = " + whichTime);
TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
Map<TopicAndPartition, PartitionOffsetRequestInfo> requestInfo = new HashMap<TopicAndPartition, PartitionOffsetRequestInfo>();
requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1));
kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(),clientName);
OffsetResponse response = consumer.getOffsetsBefore(request);
if (response.hasError()) {
System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition) + " | " + response.toString() );
return 0;
}
long[] offsets = response.offsets(topic, partition);
return offsets[0];
}
Thanks，
Lax
> Date: Fri, 22 Aug 2014 21:27:58 -0700
> Subject: Re: read message use SimpleConsumer,can not start with assign offset
> From: junrao@gmail.com
> To: users@kafka.apache.org
>
> What's code in line 164 of SimpleConsumerClient.getLastOffset()?
>
> Thanks,
>
> Jun
>
>
> On Fri, Aug 22, 2014 at 7:36 AM, chenlax <la...@hotmail.com> wrote:
>
> > kafka version is 0.8.0, the code as
> > https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
> >
> > when i assign a offset ,it will throw
> > java.lang.ArrayIndexOutOfBoundsException: 0
> >
> > java.lang.ArrayIndexOutOfBoundsException: 0
> > at
> > com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
> > at
> > com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
> > at
> > com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
> > ##EarliestTime : -2 || -1 #### The Start Offset : -2
> > Error fetching data from the Broker:10.209.22.206 Reason: 1
> >
> > how can is get the message which is assign offset?
> >
> > Thanks，
> > Lax
> >
Re: read message use SimpleConsumer,can not start with assign offset
Posted by Jun Rao <ju...@gmail.com>.
What's code in line 164 of SimpleConsumerClient.getLastOffset()?
Thanks,
Jun
On Fri, Aug 22, 2014 at 7:36 AM, chenlax <la...@hotmail.com> wrote:
> kafka version is 0.8.0, the code as
> https://cwiki.apache.org/confluence/display/KAFKA/0.8.0+SimpleConsumer+Example
>
> when i assign a offset ,it will throw
> java.lang.ArrayIndexOutOfBoundsException: 0
>
> java.lang.ArrayIndexOutOfBoundsException: 0
> at
> com.chen.test.consumer.SimpleConsumerClient.getLastOffset(SimpleConsumerClient.java:164)
> at
> com.chen.test.consumer.SimpleConsumerClient.run(SimpleConsumerClient.java:80)
> at
> com.chen.test.consumer.SimpleConsumerClient.main(SimpleConsumerClient.java:48)
> ##EarliestTime : -2 || -1 #### The Start Offset : -2
> Error fetching data from the Broker:10.209.22.206 Reason: 1
>
> how can is get the message which is assign offset?
>
> Thanks，
> Lax
>