Re: [VOTE] 1.0.0 RC0

[Vahid S Hashemian]

Hi Guozhang,

Thanks for running the release.

I tested building from source and the quickstarts on Linux, Mac, and 
Windows 64 (with Java 8 and Gradle 4.2.1).

Everything worked well on Linux and Mac, but I ran into some issues on my 
Windows 64 VM:

I reported one issue in KAFKA-6055, but it's an easy one to fix (a PR is 
already submitted).

With that fix in place I continued my testing but ran into another issue 
after build. When trying to start a broker 
(bin\windows\kafka-server-start.bat config\server.properties) I get this 
error:

[2017-10-11 21:45:11,642] FATAL  (kafka.Kafka$)
java.lang.IllegalArgumentException: Unknown signal: HUP
at sun.misc.Signal.(Unknown Source)
at kafka.Kafka$.registerHandler$1(Kafka.scala:67)
at kafka.Kafka$.registerLoggingSignalHandler(Kafka.scala:73)
at kafka.Kafka$.main(Kafka.scala:82)
at kafka.Kafka.main(Kafka.scala)

This seems to have been introduced by a recent commit (
https://github.com/apache/kafka/commit/8256f882c92daa1470382502ab94cbe2c16028f1#diff-ef81cee39236d0121040043e4d69d330
) and for some reason that fix does not work on Windows.

Thanks.
--Vahid





From:   Guozhang Wang 
To: "d...@kafka.apache.org" , 
"users@kafka.apache.org" , kafka-clients 

Date:   10/10/2017 06:34 PM
Subject:[VOTE] 1.0.0 RC0



Hello Kafka users, developers and client-developers,

This is the first candidate for release of Apache Kafka 1.0.0.

It's worth noting that starting in this version we are using a different
version protocol with three digits: *major.minor.bug-fix*

Any and all testing is welcome, but the following areas are worth
highlighting:

1. Client developers should verify that their clients can produce/consume
to/from 1.0.0 brokers (ideally with compressed and uncompressed data).
2. Performance and stress testing. Heroku and LinkedIn have helped with
this in the past (and issues have been found and fixed).
3. End users can verify that their apps work correctly with the new 
release.

This is a major version release of Apache Kafka. It includes 29 new KIPs.
See the release notes and release plan
(*https://urldefense.proofpoint.com/v2/url?u=https-3A__cwiki.apache.org_confluence_pages_viewpage.action-3FpageId-3D71764913&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj-kjJc7uSVcviKUc&m=o8kdIbhT0v8egB8hJ137dwiXmIK8WlvIwjiFebdEGwA&s=waGurzMZ-QrdW5_pNVc3hgTUFQoJ-a8786ce-ENb9UY&e=
<
https://urldefense.proofpoint.com/v2/url?u=https-3A__cwiki.apache.org_confluence_pages_viewpage.action-3FpageId-3D71764913&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj-kjJc7uSVcviKUc&m=o8kdIbhT0v8egB8hJ137dwiXmIK8WlvIwjiFebdEGwA&s=waGurzMZ-QrdW5_pNVc3hgTUFQoJ-a8786ce-ENb9UY&e=
>*)
for more details. A few feature highlights:

* Java 9 support with significantly faster TLS and CRC32C implementations
(KIP)
* JBOD improvements: disk failure only disables failed disk but not the
broker (KIP-112/KIP-113)
* Newly added metrics across all the modules (KIP-164, KIP-168, KIP-187,
KIP-188, KIP-196)
* Kafka Streams API improvements (KIP-120 / 130 / 138 / 150 / 160 / 161),
and drop compatibility "Evolving" annotations

Release notes for the 1.0.0 release:
*https://urldefense.proofpoint.com/v2/url?u=http-3A__home.apache.org_-7Eguozhang_kafka-2D1.0.0-2Drc0_RELEASE-5FNOTES.html&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj-kjJc7uSVcviKUc&m=o8kdIbhT0v8egB8hJ137dwiXmIK8WlvIwjiFebdEGwA&s=Ba5qmFWmCACG3vS4n6iTkU2tK9HtCv-YHd2YgG-B84U&e=
<
https://urldefense.proofpoint.com/v2/url?u=http-3A__home.apache.org_-7Eguozhang_kafka-2D1.0.0-2Drc0_RELEASE-5FNOTES.html&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj-kjJc7uSVcviKUc&m=o8kdIbhT0v8egB8hJ137dwiXmIK8WlvIwjiFebdEGwA&s=Ba5qmFWmCACG3vS4n6iTkU2tK9HtCv-YHd2YgG-B84U&e=
>*



*** Please download, test and vote by Friday, October 13, 8pm PT

Kafka's KEYS file containing PGP keys we use to sign the release:
https://urldefense.proofpoint.com/v2/url?u=http-3A__kafka.apache.org_KEYS&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj-kjJc7uSVcviKUc&m=o8kdIbhT0v8egB8hJ137dwiXmIK8WlvIwjiFebdEGwA&s=IN3b-XNLtG1h0LbBPS8IQUrwLnaA6ff0iJ2Xk50Nl0o&e=


* Release artifacts to be voted upon (source and binary):
*https://urldefense.proofpoint.com/v2/url?u=http-3A__home.apache.org_-7Eguozhang_kafka-2D1.0.0-2Drc0_&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj-kjJc7uSVcviKUc&m=o8kdIbhT0v8egB8hJ137dwiXmIK8WlvIwjiFebdEGwA&s=yJLBkpLm16v7BWOKBi_gqrhivUXaC2gZwInCrRSNl9s&e=
<
https://urldefense.proofpoint.com/v2/url?u=http-3A__home.apache.org_-7Eguozhang_kafka-2D1.0.0-2Drc0_&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj-kjJc7uSVcviKUc&m=o8kdIbhT0v8egB8hJ137dwiXmIK8WlvIwjiFebdEGwA&s=yJLBkpLm16v7BWOKBi_gqrhivUXaC2gZwInCrRSNl9s&e=
>*

* Maven artifacts to be voted upon:
https://urldefense.proofpoint.com/v2/url?u=https-3A__repository.apache.org_content_groups_staging_&d=DwIBaQ&c=jf_iaSHvJObTbx-siA1ZOg&r=Q_itwloTQj3_xUKl7Nzswo6KE4Nj

Re: kafka broker loosing offsets?

[Vincent Dautremont]

Hi,
We have 4 differents Kafka cluster running,
2 on 0.10.1.0
1 on 0.10.0.1
1 that was on 0.11.0.0 and last week updated to 0.11.0.1

I’ve only seen the issue happen 2 times in production usage on the cluster on 
0.11.0.0 since it’s running (about 3months).

But I’ll monitor and report it here if it ever happen again in the future. 
We’ll also upgrade all our clusters to 0.11.0.1 in the next days.

🤞🏻!

> Le 11 oct. 2017 à 17:47, Dmitriy Vsekhvalnov  a écrit 
> :
> 
> Yeah just pops up in my list. Thanks, i'll take a look.
> 
> Vincent Dautremont, if you still reading it, did you try upgrade to
> 0.11.0.1? Fixed issue?
> 
> On Wed, Oct 11, 2017 at 6:46 PM, Ben Davison 
> wrote:
> 
>> Hi Dmitriy,
>> 
>> Did you check out this thread "Incorrect consumer offsets after broker
>> restart 0.11.0.0" from Phil Luckhurst, it sounds similar.
>> 
>> Thanks,
>> 
>> Ben
>> 
>> On Wed, Oct 11, 2017 at 4:44 PM Dmitriy Vsekhvalnov <
>> dvsekhval...@gmail.com>
>> wrote:
>> 
>>> Hey, want to resurrect this thread.
>>> 
>>> Decided to do idle test, where no load data is produced to topic at all.
>>> And when we kill #101 or #102 - nothing happening. But when we kill #200
>> -
>>> consumers starts to re-consume old events from random position.
>>> 
>>> Anybody have ideas what to check?  I really expected that Kafka will fail
>>> symmetrical with respect to any broker.
>>> 
>>> On Mon, Oct 9, 2017 at 6:26 PM, Dmitriy Vsekhvalnov <
>>> dvsekhval...@gmail.com>
>>> wrote:
>>> 
 Hi tao,
 
 we had unclean leader election enabled at the beginning. But then
>>> disabled
 it and also reduced 'max.poll.records' value.  It helped little bit.
 
 But after today's testing there is strong correlation between lag spike
 and what broker we crash. For lowest ID (100) broker :
  1. it always at least 1-2 orders higher lag
  2. we start getting
 
 org.apache.kafka.clients.consumer.CommitFailedException: Commit
>> cannot be
 completed since the group has already rebalanced and assigned the
 partitions to another member. This means that the time between
>> subsequent
 calls to poll() was longer than the configured max.poll.interval.ms,
 which typically implies that the poll loop is spending too much time
 message processing. You can address this either by increasing the
>> session
 timeout or by reducing the maximum size of batches returned in poll()
>>> with
 max.poll.records.
 
  3. sometime re-consumption from random position
 
 And when we crashing other brokers (101, 102), it just lag spike of
>> ~10Ks
 order, settle down quite quickly, no consumer exceptions.
 
 Totally lost what to try next.
 
> On Sat, Oct 7, 2017 at 2:41 AM, tao xiao  wrote:
> 
> Do you have unclean leader election turned on? If killing 100 is the
>>> only
> way to reproduce the problem, it is possible with unclean leader
>>> election
> turned on that leadership was transferred to out of ISR follower which
>>> may
> not have the latest high watermark
> On Sat, Oct 7, 2017 at 3:51 AM Dmitriy Vsekhvalnov <
> dvsekhval...@gmail.com>
> wrote:
> 
>> About to verify hypothesis on monday, but looks like that in latest
> tests.
>> Need to double check.
>> 
>> On Fri, Oct 6, 2017 at 11:25 PM, Stas Chizhov 
> wrote:
>> 
>>> So no matter in what sequence you shutdown brokers it is only 1
>> that
>> causes
>>> the major problem? That would indeed be a bit weird. have you
>>> checked
>>> offsets of your consumer - right after offsets jump back - does it
> start
>>> from the topic start or does it go back to some random position?
>>> Have
> you
>>> checked if all offsets are actually being committed by consumers?
>>> 
>>> fre 6 okt. 2017 kl. 20:59 skrev Dmitriy Vsekhvalnov <
>>> dvsekhval...@gmail.com
 :
>>> 
 Yeah, probably we can dig around.
 
 One more observation, the most lag/re-consumption trouble
>>> happening
>> when
>>> we
 kill broker with lowest id (e.g. 100 from [100,101,102]).
 When crashing other brokers - there is nothing special
>> happening,
> lag
 growing little bit but nothing crazy (e.g. thousands, not
>>> millions).
 
 Is it sounds suspicious?
 
 On Fri, Oct 6, 2017 at 9:23 PM, Stas Chizhov <
>> schiz...@gmail.com>
>> wrote:
 
> Ted: when choosing earliest/latest you are saying: if it
>> happens
> that
 there
> is no "valid" offset committed for a consumer (for whatever
> reason:
> bug/misconfiguration/no luck) it will be ok to start from the
>> beginning
 or
> end of the topic. So if you are not ok with that you should
>>> choose
>>> none.
> 
> Dmitriy: Ok. Then it is spring-kafka that maintains this
>> offset
> per
> partition state for you. it m

Re: [VOTE] 1.0.0 RC0

[Guozhang Wang]

Thanks for the check Ted.

I just made the jars available at mvn staging now:

https://repository.apache.org/content/groups/staging/org/apache/kafka/


Guozhang

On Tue, Oct 10, 2017 at 6:43 PM, Ted Yu  wrote:

> Guozhang:
> I took a brief look under the staging tree.
> e.g.
> https://repository.apache.org/content/groups/staging/org/
> apache/kafka/kafka-clients/
>
> I don't see 1.0.0 jars.
>
> Would the jars be populated later ?
>
> Thanks
>
> On Tue, Oct 10, 2017 at 6:34 PM, Guozhang Wang  wrote:
>
> > Hello Kafka users, developers and client-developers,
> >
> > This is the first candidate for release of Apache Kafka 1.0.0.
> >
> > It's worth noting that starting in this version we are using a different
> > version protocol with three digits: *major.minor.bug-fix*
> >
> > Any and all testing is welcome, but the following areas are worth
> > highlighting:
> >
> > 1. Client developers should verify that their clients can produce/consume
> > to/from 1.0.0 brokers (ideally with compressed and uncompressed data).
> > 2. Performance and stress testing. Heroku and LinkedIn have helped with
> > this in the past (and issues have been found and fixed).
> > 3. End users can verify that their apps work correctly with the new
> > release.
> >
> > This is a major version release of Apache Kafka. It includes 29 new KIPs.
> > See the release notes and release plan
> > (*https://cwiki.apache.org/confluence/pages/viewpage.
> > action?pageId=71764913
> >  action?pageId=71764913
> > >*)
> > for more details. A few feature highlights:
> >
> > * Java 9 support with significantly faster TLS and CRC32C implementations
> > (KIP)
> > * JBOD improvements: disk failure only disables failed disk but not the
> > broker (KIP-112/KIP-113)
> > * Newly added metrics across all the modules (KIP-164, KIP-168, KIP-187,
> > KIP-188, KIP-196)
> > * Kafka Streams API improvements (KIP-120 / 130 / 138 / 150 / 160 / 161),
> > and drop compatibility "Evolving" annotations
> >
> > Release notes for the 1.0.0 release:
> > *http://home.apache.org/~guozhang/kafka-1.0.0-rc0/RELEASE_NOTES.html
> > *
> >
> >
> >
> > *** Please download, test and vote by Friday, October 13, 8pm PT
> >
> > Kafka's KEYS file containing PGP keys we use to sign the release:
> > http://kafka.apache.org/KEYS
> >
> > * Release artifacts to be voted upon (source and binary):
> > *http://home.apache.org/~guozhang/kafka-1.0.0-rc0/
> > *
> >
> > * Maven artifacts to be voted upon:
> > https://repository.apache.org/content/groups/staging/
> >
> > * Javadoc:
> > *http://home.apache.org/~guozhang/kafka-1.0.0-rc0/javadoc/
> > *
> >
> > * Tag to be voted upon (off 1.0 branch) is the 1.0.0-rc0 tag:
> >
> > https://git-wip-us.apache.org/repos/asf?p=kafka.git;a=tag;h=
> > 2f97bc6a9ee269bf90b019e50b4eeb43df2f1143
> >
> > * Documentation:
> > Note the documentation can't be pushed live due to changes that will not
> go
> > live until the release. You can manually verify by downloading
> > http://home.apache.org/~guozhang/kafka-1.0.0-rc0/
> > kafka_2.11-1.0.0-site-docs.tgz
> >
> > * Successful Jenkins builds for the 1.0.0 branch:
> > Unit/integration tests: https://builds.apache.org/job/kafka-1.0-jdk7/20/
> >
> >
> > /**
> >
> >
> > Thanks,
> > -- Guozhang
> >
>



-- 
-- Guozhang

Re: Getting started with stream processing

[Matthias J. Sax]

Glad it works.

If you want to use windows, what seems more natural and also allows you
to "expire old windows eventually" (with your current approach, you
never delete old window, and thus each window create a new entry in the
internal key-value store, thus, you store grows unbounded over time) it
should work out of the box, because windows automatically aligned:

https://docs.confluent.io/current/streams/developer-guide.html#tumbling-time-windows

> Tumbling time windows are aligned to the epoch, with the lower interval bound 
> being inclusive and the upper bound being exclusive. “Aligned to the epoch” 
> means that the first window starts at timestamp zero. For example, tumbling 
> windows with a size of 5000ms have predictable window boundaries 
> [0;5000),[5000;1),... — and not [1000;6000),[6000;11000),... or even 
> something “random” like [1452;6452),[6452;11452),


-Matthias



On 10/11/17 1:42 AM, RedShift wrote:
> Matthias
> 
> 
> Thanks, using grouping key of "deviceId + timestamp" with *aggregation*
> _instead of_ reducing solved it:
> 
> 
>   KGroupedStream grouped = data.groupBy(
>   (k, v) ->
>   {
>   Date dt =
> Date.from(Instant.ofEpochSecond(v.get("data").asObject().get("tss").asLong()));
> 
>   return v.get("deviceId").asString() + dateFormat.format(dt);
>   }
>   );
> 
> 
>   KTable aggregate = grouped.aggregate(
>   () -> 0,
>   (aggKey, value, aggr) -> aggr +
> value.get("data").asObject().get("load").asInt(),
>   Serdes.Integer()
>   );
> 
> I'm still trying to find out how windowing fits in. It sounds like a
> tumbling window, but a tumbling window is defined by its length. So you
> get information for the last hour that has passed, but that last hour is
> a window of NOW - 1 hour. How do I get a window to align to hours of the
> clock?
> 
> 
> 
> On 10/10/2017 19:41, Matthias J. Sax wrote:
>> Hi,
>>
>> if the aggregation returns a different type, you can use .aggregate(...)
>> instead of .reduce(...)
>>
>> Also, for you time based computation, did you consider to use windowing?
>>
>>
>> -Matthias
>>
>> On 10/10/17 6:27 AM, RedShift wrote:
>>> Hi all
>>>
>>> Complete noob with regards to stream processing, this is my first
>>> attempt. I'm going to try and explain my thought process, here's what
>>> I'm trying to do:
>>>
>>> I would like to create a sum of "load" for every hour, for every device.
>>>
>>> Incoming stream of data:
>>>
>>> {"deviceId":"1234","data":{"tss":1507619473,"load":9}}
>>> {"deviceId":"1234","data":{"tss":1507619511,"load":8}}
>>> {"deviceId":"1234","data":{"tss":1507619549,"load":5}}
>>> {"deviceId":"9876","data":{"tss":1507619587,"load":8}}
>>> {"deviceId":"1234","data":{"tss":1507619625,"load":8}}
>>> {"deviceId":"1234","data":{"tss":1507619678,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619716,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619752,"load":9}}
>>> {"deviceId":"1234","data":{"tss":1507619789,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619825,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619864,"load":8}}
>>>
>>> Where
>>> deviceId: unique ID for every device, which also doubles as the key I
>>> use
>>> tss: UNIX timestamp in seconds
>>> load: load indication
>>>
>>> Expected outcome something like this:
>>> deviceId: 1234, time: 2017-10-01 18:00, load: 25
>>> deviceId: 1234, time: 2017-10-01 19:00, load: 13
>>> deviceId: 9876, time: 2017-10-01 18:00, load: 33
>>> deviceId: 9876, time: 2017-10-01 19:00, load: 5
>>> ...
>>>
>>>
>>> So I started:
>>>
>>>    SimpleDateFormat dateFormat = new SimpleDateFormat("-MM-dd HH");
>>> // Important bit here, I use this to construct a grouping key
>>>    KStreamBuilder builder = new KStreamBuilder();
>>>    KStream data = builder.stream("telemetry");
>>>
>>> We need to group by device, so:
>>>
>>>    KGroupedStream grouped = data.groupBy((k, v) ->
>>> v.get("deviceId").asString());
>>>
>>> But now I can't group the data again by date. So I made a combined
>>> grouping key like this:
>>>
>>>    KGroupedStream grouped = data.groupBy(
>>>    (k, v) ->
>>>    {
>>>    Date dt =
>>> Date.from(Instant.ofEpochSecond(v.get("data").asObject().get("tss").asLong()));
>>>
>>>
>>>    return v.get("deviceId").asString() + dateFormat.format(dt);
>>>    }
>>>    );
>>>
>>> Now I need to reduce the groups to sum the load:
>>>
>>>    grouped.reduce(new Reducer()
>>>    {
>>>    @Override
>>>    public JsonObject apply(JsonObject v1, JsonObject v2)
>>>    {
>>>    return null;
>>>    }
>>>    });
>>>
>>> But that's a problem. I'm supposed to sum "load" here, but I also have
>>> to return a JsonObject. That doesn't seem right. So now I figure I have
>>> to extract the "load" before the reducer, but a KGroupedStream doesn't
>>> have a map() function.
>>>
>>> Back to the drawing board. So I figure let's extract the "load" and
>>> grouping key first:
>>>
>>>    KStream map = data.map(new KeyValueMappe

Kafka Consumer - org.apache.kafka.common.errors.TimeoutException: Failed to get offsets by times in 305000 ms

[SenthilKumar K]

Hi All , Recently we starting seeing Kafka Consumer error with Timeout .
What could be the cause here ?

Version : kafka_2.11-0.11.0.0

Consumer Properties:

*bootstrap.servers, enable.auto.commit,auto.commit.interval.ms
,session.timeout.ms
,group.id
,key.deserializer,value.deserializer,max.poll.records*

--Senthil

Re: Kafka cluster Error

[Kannappan, Saravanan (Contractor)]

Can you please help me to resolve this ?

Thanks
Saravanan

From: "Kannappan, Saravanan (Contractor)" 
Date: Tuesday, October 10, 2017 at 6:35 PM
To: "users@kafka.apache.org" 
Subject: Kafka cluster Error

Hello, Someone can you help me kafka server not starting after rebooting , the 
below is the error message

[2017-10-10 22:25:25,365] INFO shutting down (kafka.server.KafkaServer)
[2017-10-10 22:25:25,374] INFO shut down completed (kafka.server.KafkaServer)
[2017-10-10 22:25:25,375] FATAL Fatal error during KafkaServerStartable 
startup. Prepare to shutdown (kafka.server.KafkaServerStartable)
org.I0Itec.zkclient.exception.ZkTimeoutException: Unable to connect to 
zookeeper server 
'172.24.235.155:2181,172.24.235.162:2181,172.24.235.180:2181,172.24.235.181:2181,172.24.235.185:2181'
 with timeout of 6000 ms
at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:1233)
at org.I0Itec.zkclient.ZkClient.(ZkClient.java:157)
at org.I0Itec.zkclient.ZkClient.(ZkClient.java:131)
at 
kafka.utils.ZkUtils$.createZkClientAndConnection(ZkUtils.scala:79)
at kafka.utils.ZkUtils$.apply(ZkUtils.scala:61)
at kafka.server.KafkaServer.initZk(KafkaServer.scala:329)
at kafka.server.KafkaServer.startup(KafkaServer.scala:187)
at 
kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:39)
at kafka.Kafka$.main(Kafka.scala:67)
at kafka.Kafka.main(Kafka.scala)

Thanks
Saravanan

Re: Getting started with stream processing

[Guozhang Wang]

Regarding windowing: actually the window boundaries are aligned at epoch
(i.e. UTC 1970, 00.00.00), so the latest window is not NOW - 1 hour.


Guozhang

On Wed, Oct 11, 2017 at 1:42 AM, RedShift  wrote:

> Matthias
>
>
> Thanks, using grouping key of "deviceId + timestamp" with *aggregation*
> _instead of_ reducing solved it:
>
>
>   KGroupedStream grouped = data.groupBy(
>   (k, v) ->
>   {
>   Date dt = Date.from(Instant.ofEpochSecon
> d(v.get("data").asObject().get("tss").asLong()));
>   return v.get("deviceId").asString() + dateFormat.format(dt);
>   }
>   );
>
>
>   KTable aggregate = grouped.aggregate(
>   () -> 0,
>   (aggKey, value, aggr) -> aggr + value.get("data").asObject().g
> et("load").asInt(),
>   Serdes.Integer()
>   );
>
> I'm still trying to find out how windowing fits in. It sounds like a
> tumbling window, but a tumbling window is defined by its length. So you get
> information for the last hour that has passed, but that last hour is a
> window of NOW - 1 hour. How do I get a window to align to hours of the
> clock?
>
>
>
>
> On 10/10/2017 19:41, Matthias J. Sax wrote:
>
>> Hi,
>>
>> if the aggregation returns a different type, you can use .aggregate(...)
>> instead of .reduce(...)
>>
>> Also, for you time based computation, did you consider to use windowing?
>>
>>
>> -Matthias
>>
>> On 10/10/17 6:27 AM, RedShift wrote:
>>
>>> Hi all
>>>
>>> Complete noob with regards to stream processing, this is my first
>>> attempt. I'm going to try and explain my thought process, here's what
>>> I'm trying to do:
>>>
>>> I would like to create a sum of "load" for every hour, for every device.
>>>
>>> Incoming stream of data:
>>>
>>> {"deviceId":"1234","data":{"tss":1507619473,"load":9}}
>>> {"deviceId":"1234","data":{"tss":1507619511,"load":8}}
>>> {"deviceId":"1234","data":{"tss":1507619549,"load":5}}
>>> {"deviceId":"9876","data":{"tss":1507619587,"load":8}}
>>> {"deviceId":"1234","data":{"tss":1507619625,"load":8}}
>>> {"deviceId":"1234","data":{"tss":1507619678,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619716,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619752,"load":9}}
>>> {"deviceId":"1234","data":{"tss":1507619789,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619825,"load":8}}
>>> {"deviceId":"9876","data":{"tss":1507619864,"load":8}}
>>>
>>> Where
>>> deviceId: unique ID for every device, which also doubles as the key I use
>>> tss: UNIX timestamp in seconds
>>> load: load indication
>>>
>>> Expected outcome something like this:
>>> deviceId: 1234, time: 2017-10-01 18:00, load: 25
>>> deviceId: 1234, time: 2017-10-01 19:00, load: 13
>>> deviceId: 9876, time: 2017-10-01 18:00, load: 33
>>> deviceId: 9876, time: 2017-10-01 19:00, load: 5
>>> ...
>>>
>>>
>>> So I started:
>>>
>>>SimpleDateFormat dateFormat = new SimpleDateFormat("-MM-dd HH");
>>> // Important bit here, I use this to construct a grouping key
>>>KStreamBuilder builder = new KStreamBuilder();
>>>KStream data = builder.stream("telemetry");
>>>
>>> We need to group by device, so:
>>>
>>>KGroupedStream grouped = data.groupBy((k, v) ->
>>> v.get("deviceId").asString());
>>>
>>> But now I can't group the data again by date. So I made a combined
>>> grouping key like this:
>>>
>>>KGroupedStream grouped = data.groupBy(
>>>(k, v) ->
>>>{
>>>Date dt =
>>> Date.from(Instant.ofEpochSecond(v.get("data").asObject().
>>> get("tss").asLong()));
>>>
>>>return v.get("deviceId").asString() + dateFormat.format(dt);
>>>}
>>>);
>>>
>>> Now I need to reduce the groups to sum the load:
>>>
>>>grouped.reduce(new Reducer()
>>>{
>>>@Override
>>>public JsonObject apply(JsonObject v1, JsonObject v2)
>>>{
>>>return null;
>>>}
>>>});
>>>
>>> But that's a problem. I'm supposed to sum "load" here, but I also have
>>> to return a JsonObject. That doesn't seem right. So now I figure I have
>>> to extract the "load" before the reducer, but a KGroupedStream doesn't
>>> have a map() function.
>>>
>>> Back to the drawing board. So I figure let's extract the "load" and
>>> grouping key first:
>>>
>>>KStream map = data.map(new KeyValueMapper>> JsonObject, KeyValue>()
>>>{
>>>@Override
>>>public KeyValue apply(String s, JsonObject v)
>>>{
>>>Date dt =
>>> Date.from(Instant.ofEpochSecond(v.get("data").asObject().
>>> get("tss").asLong()));
>>>
>>>String key = v.get("deviceId").asString() +
>>> dateFormat.format(dt);
>>>
>>>return new KeyValue<>(
>>>key, v.get("data").asObject().get("load").asInt()
>>>);
>>>}
>>>});
>>>
>>> But now I'm left with a KStream of . I've lost my types.
>>> If I change it to:
>>> Kstream, the compiler has this to say:
>>>
>>> Error:(35, 54) java: incompatible types: inference variable KR has
>>> incompatible bounds
>>>  equality constraints: j

Re: kafka broker loosing offsets?

[Michal Michalski]

Hi Dmitriy,

I didn't follow the whole thread, but if it's not an issue with Kafka
0.11.0.0 (there was another thread about it recently), make sure your
Replication Factor for the offsets topic is 3 (you mentioned "RF=3 for all
topics", but I wasn't sure it includes the offsets one).

There was a bug in older Kafka versions [1] that should be fixed already in
0.11, but *if* your offset topic was created earlier (e.g. you were running
older Kafka version that you only recently upgraded) it might be not
replicated as you'd expect it to be (RF=1 rather than 3) *and* if you're
using ephemeral storage (e.g. AWS EC2 instance storage), it would mean that
restarting node wipes out offset data you're looking for, so you always
start from scratch. It sounds like an unlikely scenario that would require
some very specific preconditions and a bit of bad luck, but if you check
everything else and run out of other ideas, maybe it's worth checking this
possibility as well :-)

[1] https://issues.apache.org/jira/browse/KAFKA-3959

Michal


On 11 October 2017 at 16:44, Dmitriy Vsekhvalnov 
wrote:

> Hey, want to resurrect this thread.
>
> Decided to do idle test, where no load data is produced to topic at all.
> And when we kill #101 or #102 - nothing happening. But when we kill #200 -
> consumers starts to re-consume old events from random position.
>
> Anybody have ideas what to check?  I really expected that Kafka will fail
> symmetrical with respect to any broker.
>
> On Mon, Oct 9, 2017 at 6:26 PM, Dmitriy Vsekhvalnov <
> dvsekhval...@gmail.com>
> wrote:
>
> > Hi tao,
> >
> > we had unclean leader election enabled at the beginning. But then
> disabled
> > it and also reduced 'max.poll.records' value.  It helped little bit.
> >
> > But after today's testing there is strong correlation between lag spike
> > and what broker we crash. For lowest ID (100) broker :
> >   1. it always at least 1-2 orders higher lag
> >   2. we start getting
> >
> > org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot
> be
> > completed since the group has already rebalanced and assigned the
> > partitions to another member. This means that the time between subsequent
> > calls to poll() was longer than the configured max.poll.interval.ms,
> > which typically implies that the poll loop is spending too much time
> > message processing. You can address this either by increasing the session
> > timeout or by reducing the maximum size of batches returned in poll()
> with
> > max.poll.records.
> >
> >   3. sometime re-consumption from random position
> >
> > And when we crashing other brokers (101, 102), it just lag spike of ~10Ks
> > order, settle down quite quickly, no consumer exceptions.
> >
> > Totally lost what to try next.
> >
> > On Sat, Oct 7, 2017 at 2:41 AM, tao xiao  wrote:
> >
> >> Do you have unclean leader election turned on? If killing 100 is the
> only
> >> way to reproduce the problem, it is possible with unclean leader
> election
> >> turned on that leadership was transferred to out of ISR follower which
> may
> >> not have the latest high watermark
> >> On Sat, Oct 7, 2017 at 3:51 AM Dmitriy Vsekhvalnov <
> >> dvsekhval...@gmail.com>
> >> wrote:
> >>
> >> > About to verify hypothesis on monday, but looks like that in latest
> >> tests.
> >> > Need to double check.
> >> >
> >> > On Fri, Oct 6, 2017 at 11:25 PM, Stas Chizhov 
> >> wrote:
> >> >
> >> > > So no matter in what sequence you shutdown brokers it is only 1 that
> >> > causes
> >> > > the major problem? That would indeed be a bit weird. have you
> checked
> >> > > offsets of your consumer - right after offsets jump back - does it
> >> start
> >> > > from the topic start or does it go back to some random position?
> Have
> >> you
> >> > > checked if all offsets are actually being committed by consumers?
> >> > >
> >> > > fre 6 okt. 2017 kl. 20:59 skrev Dmitriy Vsekhvalnov <
> >> > > dvsekhval...@gmail.com
> >> > > >:
> >> > >
> >> > > > Yeah, probably we can dig around.
> >> > > >
> >> > > > One more observation, the most lag/re-consumption trouble
> happening
> >> > when
> >> > > we
> >> > > > kill broker with lowest id (e.g. 100 from [100,101,102]).
> >> > > > When crashing other brokers - there is nothing special happening,
> >> lag
> >> > > > growing little bit but nothing crazy (e.g. thousands, not
> millions).
> >> > > >
> >> > > > Is it sounds suspicious?
> >> > > >
> >> > > > On Fri, Oct 6, 2017 at 9:23 PM, Stas Chizhov 
> >> > wrote:
> >> > > >
> >> > > > > Ted: when choosing earliest/latest you are saying: if it happens
> >> that
> >> > > > there
> >> > > > > is no "valid" offset committed for a consumer (for whatever
> >> reason:
> >> > > > > bug/misconfiguration/no luck) it will be ok to start from the
> >> > beginning
> >> > > > or
> >> > > > > end of the topic. So if you are not ok with that you should
> choose
> >> > > none.
> >> > > > >
> >> > > > > Dmitriy: Ok. Then it is spring-kafka that maintains this offset
> >> per
> >> > > > > parti

Retrieve unacknowledged messages from partition

[Tarik Courdy]

Good morning -

First off, thank you for apache kafka.  I have been having fun learning it
and getting it set up.

I do have a couple of questions that I haven't been able to find the answer
to yet that I'm hoping some can help with.

1.) Is there a programmatic API way to retrieve the list of active
consumers on a partition?

2.) Is there an API for retrieving messages that have been placed on a
partition but have not yet been acknowledged by any consumer?

Thank you very much for your time.

-Tarik

Re: kafka broker loosing offsets?

[Dmitriy Vsekhvalnov]

Yeah just pops up in my list. Thanks, i'll take a look.

Vincent Dautremont, if you still reading it, did you try upgrade to
0.11.0.1? Fixed issue?

On Wed, Oct 11, 2017 at 6:46 PM, Ben Davison 
wrote:

> Hi Dmitriy,
>
> Did you check out this thread "Incorrect consumer offsets after broker
> restart 0.11.0.0" from Phil Luckhurst, it sounds similar.
>
> Thanks,
>
> Ben
>
> On Wed, Oct 11, 2017 at 4:44 PM Dmitriy Vsekhvalnov <
> dvsekhval...@gmail.com>
> wrote:
>
> > Hey, want to resurrect this thread.
> >
> > Decided to do idle test, where no load data is produced to topic at all.
> > And when we kill #101 or #102 - nothing happening. But when we kill #200
> -
> > consumers starts to re-consume old events from random position.
> >
> > Anybody have ideas what to check?  I really expected that Kafka will fail
> > symmetrical with respect to any broker.
> >
> > On Mon, Oct 9, 2017 at 6:26 PM, Dmitriy Vsekhvalnov <
> > dvsekhval...@gmail.com>
> > wrote:
> >
> > > Hi tao,
> > >
> > > we had unclean leader election enabled at the beginning. But then
> > disabled
> > > it and also reduced 'max.poll.records' value.  It helped little bit.
> > >
> > > But after today's testing there is strong correlation between lag spike
> > > and what broker we crash. For lowest ID (100) broker :
> > >   1. it always at least 1-2 orders higher lag
> > >   2. we start getting
> > >
> > > org.apache.kafka.clients.consumer.CommitFailedException: Commit
> cannot be
> > > completed since the group has already rebalanced and assigned the
> > > partitions to another member. This means that the time between
> subsequent
> > > calls to poll() was longer than the configured max.poll.interval.ms,
> > > which typically implies that the poll loop is spending too much time
> > > message processing. You can address this either by increasing the
> session
> > > timeout or by reducing the maximum size of batches returned in poll()
> > with
> > > max.poll.records.
> > >
> > >   3. sometime re-consumption from random position
> > >
> > > And when we crashing other brokers (101, 102), it just lag spike of
> ~10Ks
> > > order, settle down quite quickly, no consumer exceptions.
> > >
> > > Totally lost what to try next.
> > >
> > > On Sat, Oct 7, 2017 at 2:41 AM, tao xiao  wrote:
> > >
> > >> Do you have unclean leader election turned on? If killing 100 is the
> > only
> > >> way to reproduce the problem, it is possible with unclean leader
> > election
> > >> turned on that leadership was transferred to out of ISR follower which
> > may
> > >> not have the latest high watermark
> > >> On Sat, Oct 7, 2017 at 3:51 AM Dmitriy Vsekhvalnov <
> > >> dvsekhval...@gmail.com>
> > >> wrote:
> > >>
> > >> > About to verify hypothesis on monday, but looks like that in latest
> > >> tests.
> > >> > Need to double check.
> > >> >
> > >> > On Fri, Oct 6, 2017 at 11:25 PM, Stas Chizhov 
> > >> wrote:
> > >> >
> > >> > > So no matter in what sequence you shutdown brokers it is only 1
> that
> > >> > causes
> > >> > > the major problem? That would indeed be a bit weird. have you
> > checked
> > >> > > offsets of your consumer - right after offsets jump back - does it
> > >> start
> > >> > > from the topic start or does it go back to some random position?
> > Have
> > >> you
> > >> > > checked if all offsets are actually being committed by consumers?
> > >> > >
> > >> > > fre 6 okt. 2017 kl. 20:59 skrev Dmitriy Vsekhvalnov <
> > >> > > dvsekhval...@gmail.com
> > >> > > >:
> > >> > >
> > >> > > > Yeah, probably we can dig around.
> > >> > > >
> > >> > > > One more observation, the most lag/re-consumption trouble
> > happening
> > >> > when
> > >> > > we
> > >> > > > kill broker with lowest id (e.g. 100 from [100,101,102]).
> > >> > > > When crashing other brokers - there is nothing special
> happening,
> > >> lag
> > >> > > > growing little bit but nothing crazy (e.g. thousands, not
> > millions).
> > >> > > >
> > >> > > > Is it sounds suspicious?
> > >> > > >
> > >> > > > On Fri, Oct 6, 2017 at 9:23 PM, Stas Chizhov <
> schiz...@gmail.com>
> > >> > wrote:
> > >> > > >
> > >> > > > > Ted: when choosing earliest/latest you are saying: if it
> happens
> > >> that
> > >> > > > there
> > >> > > > > is no "valid" offset committed for a consumer (for whatever
> > >> reason:
> > >> > > > > bug/misconfiguration/no luck) it will be ok to start from the
> > >> > beginning
> > >> > > > or
> > >> > > > > end of the topic. So if you are not ok with that you should
> > choose
> > >> > > none.
> > >> > > > >
> > >> > > > > Dmitriy: Ok. Then it is spring-kafka that maintains this
> offset
> > >> per
> > >> > > > > partition state for you. it might also has that problem of
> > leaving
> > >> > > stale
> > >> > > > > offsets lying around, After quickly looking through
> > >> > > > > https://github.com/spring-projects/spring-kafka/blob/
> > >> > > > > 1945f29d5518e3c4a9950ba82135420dfb61e808/spring-kafka/src/
> > >> > > > > main/java/org/springframework/kafka/listener/
> > >> > > > > KafkaM

Re: kafka broker loosing offsets?

[Ben Davison]

Hi Dmitriy,

Did you check out this thread "Incorrect consumer offsets after broker
restart 0.11.0.0" from Phil Luckhurst, it sounds similar.

Thanks,

Ben

On Wed, Oct 11, 2017 at 4:44 PM Dmitriy Vsekhvalnov 
wrote:

> Hey, want to resurrect this thread.
>
> Decided to do idle test, where no load data is produced to topic at all.
> And when we kill #101 or #102 - nothing happening. But when we kill #200 -
> consumers starts to re-consume old events from random position.
>
> Anybody have ideas what to check?  I really expected that Kafka will fail
> symmetrical with respect to any broker.
>
> On Mon, Oct 9, 2017 at 6:26 PM, Dmitriy Vsekhvalnov <
> dvsekhval...@gmail.com>
> wrote:
>
> > Hi tao,
> >
> > we had unclean leader election enabled at the beginning. But then
> disabled
> > it and also reduced 'max.poll.records' value.  It helped little bit.
> >
> > But after today's testing there is strong correlation between lag spike
> > and what broker we crash. For lowest ID (100) broker :
> >   1. it always at least 1-2 orders higher lag
> >   2. we start getting
> >
> > org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be
> > completed since the group has already rebalanced and assigned the
> > partitions to another member. This means that the time between subsequent
> > calls to poll() was longer than the configured max.poll.interval.ms,
> > which typically implies that the poll loop is spending too much time
> > message processing. You can address this either by increasing the session
> > timeout or by reducing the maximum size of batches returned in poll()
> with
> > max.poll.records.
> >
> >   3. sometime re-consumption from random position
> >
> > And when we crashing other brokers (101, 102), it just lag spike of ~10Ks
> > order, settle down quite quickly, no consumer exceptions.
> >
> > Totally lost what to try next.
> >
> > On Sat, Oct 7, 2017 at 2:41 AM, tao xiao  wrote:
> >
> >> Do you have unclean leader election turned on? If killing 100 is the
> only
> >> way to reproduce the problem, it is possible with unclean leader
> election
> >> turned on that leadership was transferred to out of ISR follower which
> may
> >> not have the latest high watermark
> >> On Sat, Oct 7, 2017 at 3:51 AM Dmitriy Vsekhvalnov <
> >> dvsekhval...@gmail.com>
> >> wrote:
> >>
> >> > About to verify hypothesis on monday, but looks like that in latest
> >> tests.
> >> > Need to double check.
> >> >
> >> > On Fri, Oct 6, 2017 at 11:25 PM, Stas Chizhov 
> >> wrote:
> >> >
> >> > > So no matter in what sequence you shutdown brokers it is only 1 that
> >> > causes
> >> > > the major problem? That would indeed be a bit weird. have you
> checked
> >> > > offsets of your consumer - right after offsets jump back - does it
> >> start
> >> > > from the topic start or does it go back to some random position?
> Have
> >> you
> >> > > checked if all offsets are actually being committed by consumers?
> >> > >
> >> > > fre 6 okt. 2017 kl. 20:59 skrev Dmitriy Vsekhvalnov <
> >> > > dvsekhval...@gmail.com
> >> > > >:
> >> > >
> >> > > > Yeah, probably we can dig around.
> >> > > >
> >> > > > One more observation, the most lag/re-consumption trouble
> happening
> >> > when
> >> > > we
> >> > > > kill broker with lowest id (e.g. 100 from [100,101,102]).
> >> > > > When crashing other brokers - there is nothing special happening,
> >> lag
> >> > > > growing little bit but nothing crazy (e.g. thousands, not
> millions).
> >> > > >
> >> > > > Is it sounds suspicious?
> >> > > >
> >> > > > On Fri, Oct 6, 2017 at 9:23 PM, Stas Chizhov 
> >> > wrote:
> >> > > >
> >> > > > > Ted: when choosing earliest/latest you are saying: if it happens
> >> that
> >> > > > there
> >> > > > > is no "valid" offset committed for a consumer (for whatever
> >> reason:
> >> > > > > bug/misconfiguration/no luck) it will be ok to start from the
> >> > beginning
> >> > > > or
> >> > > > > end of the topic. So if you are not ok with that you should
> choose
> >> > > none.
> >> > > > >
> >> > > > > Dmitriy: Ok. Then it is spring-kafka that maintains this offset
> >> per
> >> > > > > partition state for you. it might also has that problem of
> leaving
> >> > > stale
> >> > > > > offsets lying around, After quickly looking through
> >> > > > > https://github.com/spring-projects/spring-kafka/blob/
> >> > > > > 1945f29d5518e3c4a9950ba82135420dfb61e808/spring-kafka/src/
> >> > > > > main/java/org/springframework/kafka/listener/
> >> > > > > KafkaMessageListenerContainer.java
> >> > > > > it looks possible since offsets map is not cleared upon
> partition
> >> > > > > revocation, but that is just a hypothesis. I have no experience
> >> with
> >> > > > > spring-kafka. However since you say you consumers were always
> >> active
> >> > I
> >> > > > find
> >> > > > > this theory worth investigating.
> >> > > > >
> >> > > > >
> >> > > > > 2017-10-06 18:20 GMT+02:00 Vincent Dautremont <
> >> > > > > vincent.dautrem...@olamobile.com.invalid>:
> >> > > > >
> >> > > 

Re: kafka broker loosing offsets?

[Dmitriy Vsekhvalnov]

Hey, want to resurrect this thread.

Decided to do idle test, where no load data is produced to topic at all.
And when we kill #101 or #102 - nothing happening. But when we kill #200 -
consumers starts to re-consume old events from random position.

Anybody have ideas what to check?  I really expected that Kafka will fail
symmetrical with respect to any broker.

On Mon, Oct 9, 2017 at 6:26 PM, Dmitriy Vsekhvalnov 
wrote:

> Hi tao,
>
> we had unclean leader election enabled at the beginning. But then disabled
> it and also reduced 'max.poll.records' value.  It helped little bit.
>
> But after today's testing there is strong correlation between lag spike
> and what broker we crash. For lowest ID (100) broker :
>   1. it always at least 1-2 orders higher lag
>   2. we start getting
>
> org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be
> completed since the group has already rebalanced and assigned the
> partitions to another member. This means that the time between subsequent
> calls to poll() was longer than the configured max.poll.interval.ms,
> which typically implies that the poll loop is spending too much time
> message processing. You can address this either by increasing the session
> timeout or by reducing the maximum size of batches returned in poll() with
> max.poll.records.
>
>   3. sometime re-consumption from random position
>
> And when we crashing other brokers (101, 102), it just lag spike of ~10Ks
> order, settle down quite quickly, no consumer exceptions.
>
> Totally lost what to try next.
>
> On Sat, Oct 7, 2017 at 2:41 AM, tao xiao  wrote:
>
>> Do you have unclean leader election turned on? If killing 100 is the only
>> way to reproduce the problem, it is possible with unclean leader election
>> turned on that leadership was transferred to out of ISR follower which may
>> not have the latest high watermark
>> On Sat, Oct 7, 2017 at 3:51 AM Dmitriy Vsekhvalnov <
>> dvsekhval...@gmail.com>
>> wrote:
>>
>> > About to verify hypothesis on monday, but looks like that in latest
>> tests.
>> > Need to double check.
>> >
>> > On Fri, Oct 6, 2017 at 11:25 PM, Stas Chizhov 
>> wrote:
>> >
>> > > So no matter in what sequence you shutdown brokers it is only 1 that
>> > causes
>> > > the major problem? That would indeed be a bit weird. have you checked
>> > > offsets of your consumer - right after offsets jump back - does it
>> start
>> > > from the topic start or does it go back to some random position? Have
>> you
>> > > checked if all offsets are actually being committed by consumers?
>> > >
>> > > fre 6 okt. 2017 kl. 20:59 skrev Dmitriy Vsekhvalnov <
>> > > dvsekhval...@gmail.com
>> > > >:
>> > >
>> > > > Yeah, probably we can dig around.
>> > > >
>> > > > One more observation, the most lag/re-consumption trouble happening
>> > when
>> > > we
>> > > > kill broker with lowest id (e.g. 100 from [100,101,102]).
>> > > > When crashing other brokers - there is nothing special happening,
>> lag
>> > > > growing little bit but nothing crazy (e.g. thousands, not millions).
>> > > >
>> > > > Is it sounds suspicious?
>> > > >
>> > > > On Fri, Oct 6, 2017 at 9:23 PM, Stas Chizhov 
>> > wrote:
>> > > >
>> > > > > Ted: when choosing earliest/latest you are saying: if it happens
>> that
>> > > > there
>> > > > > is no "valid" offset committed for a consumer (for whatever
>> reason:
>> > > > > bug/misconfiguration/no luck) it will be ok to start from the
>> > beginning
>> > > > or
>> > > > > end of the topic. So if you are not ok with that you should choose
>> > > none.
>> > > > >
>> > > > > Dmitriy: Ok. Then it is spring-kafka that maintains this offset
>> per
>> > > > > partition state for you. it might also has that problem of leaving
>> > > stale
>> > > > > offsets lying around, After quickly looking through
>> > > > > https://github.com/spring-projects/spring-kafka/blob/
>> > > > > 1945f29d5518e3c4a9950ba82135420dfb61e808/spring-kafka/src/
>> > > > > main/java/org/springframework/kafka/listener/
>> > > > > KafkaMessageListenerContainer.java
>> > > > > it looks possible since offsets map is not cleared upon partition
>> > > > > revocation, but that is just a hypothesis. I have no experience
>> with
>> > > > > spring-kafka. However since you say you consumers were always
>> active
>> > I
>> > > > find
>> > > > > this theory worth investigating.
>> > > > >
>> > > > >
>> > > > > 2017-10-06 18:20 GMT+02:00 Vincent Dautremont <
>> > > > > vincent.dautrem...@olamobile.com.invalid>:
>> > > > >
>> > > > > > is there a way to read messages on a topic partition from a
>> > specific
>> > > > node
>> > > > > > we that we choose (and not by the topic partition leader) ?
>> > > > > > I would like to read myself that each of the __consumer_offsets
>> > > > partition
>> > > > > > replicas have the same consumer group offset written in it in
>> it.
>> > > > > >
>> > > > > > On Fri, Oct 6, 2017 at 6:08 PM, Dmitriy Vsekhvalnov <
>> > > > > > dvsekhval...@gmail.com>
>> > > > > > wrote:
>> > > > > >
>> >

Questions about Apache Kafka messages type/size and publish/subscribe

[Heloise Chevalier]

Hi,

I'm not entirely certain this is the right place to ask, but I have
questions about the functioning of Apache Kafka to implement a
publish/subscribe messaging system. I am investigating Kafka to see if it
fits the needs of the company I work for, and I have quite a few questions
I can't find any answer to either in the official doc or on the internet :

On the type and size of Kafka messages, I read the doc on RecordBatch and
Records, but I couldn't find anything about the type of data expected in
the Record value. From what I read on different websites and tutorials it
seems that data can be anything as long as it can be converted into byte
arrays. Moreover, I read that the default size of a message is 1MB, but as
it is possible to configure it, what would be the maximum size of a
message?

- Is it possible to send files? Basically what we would want to do is
streaming pictures, right now the system in use does not allow to send
files so they are sent as binary data and assembled later. From what I
understand of Kafka, files can be sent in a binary way, split and to be
reassembled after.

- Is it possible to send binary frames? Right now we have a few "custom"
binary frame and we would like to know if this would still be possible when
using Kafka.

- Is it possible to send simple messages, such as a string + an int or a
string + a string...?

- What is the max size of a queue, if there is one? We would like to know
what is the max volume of data that can be sent instantaneously (depending
on the size of the queue, I suppose).

On publish-subscribe more specifically :

- Is it possible for a consumer to subscribe to a group of publishers?
Otherwise I guess the easiest way to do so is to have all publishers from
that group publish their messages in a specific topic.

- Is it possible to subscribe to a group of topics? For example if we have
several topics starting by "data", like data_1, data_2 etc, would it be
possible to have consumers subscribe to "data*" ?

- Is it possible to have sub-topics, and even sub-sub-sopics? From what I
understand so far it doesn't seem possible, as topics are divided into
partitions for replication.



I know this is a lot of questions and that they are not really related. I
hope I made myself clear enough and I thank you in advance for your answers.

Regards,

Héloïse Chevalier

Re: Incorrect consumer offsets after broker restart 0.11.0.0

[Vincent Dautremont]

I would also like to know the related Jira ticket if any, to check that
what I experience the same phenomenon.
I see this happening even without restarting the kafka broker process :

I sometime have a Zookeeper socket that fails, the Kafka broker then step
down from its leader duties for a few seconds before entering the cluster,
Consumers gets a response from the broker that it is not anymore part of
the cluster.

So my client resets, gets reassigned partitions and gets some old CG
offsets from day ago (often from log data that was already deleted).

Thanks.


On Wed, Oct 11, 2017 at 1:59 PM, Phil Luckhurst 
wrote:

> Upgrading the broker to version 0.11.0.1 has fixed the problem.
>
> Thanks.
>

-- 
The information transmitted is intended only for the person or entity to 
which it is addressed and may contain confidential and/or privileged 
material. Any review, retransmission, dissemination or other use of, or 
taking of any action in reliance upon, this information by persons or 
entities other than the intended recipient is prohibited. If you received 
this in error, please contact the sender and delete the material from any 
computer.

RE: Incorrect consumer offsets after broker restart 0.11.0.0

[Phil Luckhurst]

Upgrading the broker to version 0.11.0.1 has fixed the problem.

Thanks.

Re: Getting started with stream processing

[RedShift]

Matthias


Thanks, using grouping key of "deviceId + timestamp" with *aggregation* 
_instead of_ reducing solved it:


  KGroupedStream grouped = data.groupBy(
  (k, v) ->
  {
  Date dt = 
Date.from(Instant.ofEpochSecond(v.get("data").asObject().get("tss").asLong()));
  return v.get("deviceId").asString() + dateFormat.format(dt);
  }
  );


  KTable aggregate = grouped.aggregate(
  () -> 0,
  (aggKey, value, aggr) -> aggr + 
value.get("data").asObject().get("load").asInt(),
  Serdes.Integer()
  );

I'm still trying to find out how windowing fits in. It sounds like a tumbling 
window, but a tumbling window is defined by its length. So you get information 
for the last hour that has passed, but that last hour is a window of NOW - 1 
hour. How do I get a window to align to hours of the clock?



On 10/10/2017 19:41, Matthias J. Sax wrote:

Hi,

if the aggregation returns a different type, you can use .aggregate(...)
instead of .reduce(...)

Also, for you time based computation, did you consider to use windowing?


-Matthias

On 10/10/17 6:27 AM, RedShift wrote:

Hi all

Complete noob with regards to stream processing, this is my first
attempt. I'm going to try and explain my thought process, here's what
I'm trying to do:

I would like to create a sum of "load" for every hour, for every device.

Incoming stream of data:

{"deviceId":"1234","data":{"tss":1507619473,"load":9}}
{"deviceId":"1234","data":{"tss":1507619511,"load":8}}
{"deviceId":"1234","data":{"tss":1507619549,"load":5}}
{"deviceId":"9876","data":{"tss":1507619587,"load":8}}
{"deviceId":"1234","data":{"tss":1507619625,"load":8}}
{"deviceId":"1234","data":{"tss":1507619678,"load":8}}
{"deviceId":"9876","data":{"tss":1507619716,"load":8}}
{"deviceId":"9876","data":{"tss":1507619752,"load":9}}
{"deviceId":"1234","data":{"tss":1507619789,"load":8}}
{"deviceId":"9876","data":{"tss":1507619825,"load":8}}
{"deviceId":"9876","data":{"tss":1507619864,"load":8}}

Where
deviceId: unique ID for every device, which also doubles as the key I use
tss: UNIX timestamp in seconds
load: load indication

Expected outcome something like this:
deviceId: 1234, time: 2017-10-01 18:00, load: 25
deviceId: 1234, time: 2017-10-01 19:00, load: 13
deviceId: 9876, time: 2017-10-01 18:00, load: 33
deviceId: 9876, time: 2017-10-01 19:00, load: 5
...


So I started:

   SimpleDateFormat dateFormat = new SimpleDateFormat("-MM-dd HH");
// Important bit here, I use this to construct a grouping key
   KStreamBuilder builder = new KStreamBuilder();
   KStream data = builder.stream("telemetry");

We need to group by device, so:

   KGroupedStream grouped = data.groupBy((k, v) ->
v.get("deviceId").asString());

But now I can't group the data again by date. So I made a combined
grouping key like this:

   KGroupedStream grouped = data.groupBy(
   (k, v) ->
   {
   Date dt =
Date.from(Instant.ofEpochSecond(v.get("data").asObject().get("tss").asLong()));

   return v.get("deviceId").asString() + dateFormat.format(dt);
   }
   );

Now I need to reduce the groups to sum the load:

   grouped.reduce(new Reducer()
   {
   @Override
   public JsonObject apply(JsonObject v1, JsonObject v2)
   {
   return null;
   }
   });

But that's a problem. I'm supposed to sum "load" here, but I also have
to return a JsonObject. That doesn't seem right. So now I figure I have
to extract the "load" before the reducer, but a KGroupedStream doesn't
have a map() function.

Back to the drawing board. So I figure let's extract the "load" and
grouping key first:

   KStream map = data.map(new KeyValueMapper>()
   {
   @Override
   public KeyValue apply(String s, JsonObject v)
   {
   Date dt =
Date.from(Instant.ofEpochSecond(v.get("data").asObject().get("tss").asLong()));

   String key = v.get("deviceId").asString() +
dateFormat.format(dt);

   return new KeyValue<>(
   key, v.get("data").asObject().get("load").asInt()
   );
   }
   });

But now I'm left with a KStream of . I've lost my types.
If I change it to:
Kstream, the compiler has this to say:

Error:(35, 54) java: incompatible types: inference variable KR has
incompatible bounds
     equality constraints: java.lang.String
     lower bounds: java.lang.Object
     Makes sense, as there's no garantuee that a random given object is a
string. But how do I preserve types then?

I'm also unsure about the way I'm grouping things. It seems to me I have
to group by deviceId, and then using windowing to get the "per hour"
part. But I'm even more clueless how and where that fits in. For some
reason I also think a KTable should be the final result?

Thanks,

Best regards,

Re: Add Kafka user list

[Jaikiran Pai]

I'm curious how these emails even get delivered to this (and sometimes 
the dev list) if the user isn't yet subscribed (through the 
users-subscribe mailing list)? Is this mailing list setup to accept 
mails from unsubscribed users?


-Jaikiran


On 11/10/17 12:32 PM, Jakub Scholz wrote:

Out of curiosity ... there seem to be quite a lot of these emails. I wonder
if we can do something to improve on this.

Was someone thinking about changing the UX on the Kafka website? Maybe
removing the links from the users@kafka... email? Or rephrasing the
sentence to make the subscribe email be first in the sentence? Or at least
making the subscribe links bold?

Also, when I first came to the Kafka website to look for the mailing lists,
the "Contact Us" section wasn't my first choice to look for them. Wouldn't
renaming it to "Discussion" or "Community" be more intuitive?

Thanks
Jakub

On Tue, Oct 10, 2017 at 7:35 PM, Matthias J. Sax 
wrote:


If you want to subscribe follow instructions here:
http://kafka.apache.org/contact

On 10/10/17 2:07 AM, shawnding(丁晓坤) wrote:

Add Kafka user list

Re: Add Kafka user list

[Jakub Scholz]

Out of curiosity ... there seem to be quite a lot of these emails. I wonder
if we can do something to improve on this.

Was someone thinking about changing the UX on the Kafka website? Maybe
removing the links from the users@kafka... email? Or rephrasing the
sentence to make the subscribe email be first in the sentence? Or at least
making the subscribe links bold?

Also, when I first came to the Kafka website to look for the mailing lists,
the "Contact Us" section wasn't my first choice to look for them. Wouldn't
renaming it to "Discussion" or "Community" be more intuitive?

Thanks
Jakub

On Tue, Oct 10, 2017 at 7:35 PM, Matthias J. Sax 
wrote:

> If you want to subscribe follow instructions here:
> http://kafka.apache.org/contact
>
> On 10/10/17 2:07 AM, shawnding(丁晓坤) wrote:
> > Add Kafka user list
> >
>
>