[GitHub] kafka pull request #3333: MINOR: Mark AbstractLogCleanerIntegrationTest as a...

[ewencp]

GitHub user ewencp opened a pull request:

https://github.com/apache/kafka/pull/

MINOR: Mark AbstractLogCleanerIntegrationTest as an IntegrationTest



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/ewencp/kafka 
minor-mark-log-cleaner-integration-test

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #


commit ddbce35d52ee42b30334219c8e84829c6bc19d6a
Author: Ewen Cheslack-Postava 
Date:   2017-06-14T06:22:53Z

MINOR: Mark AbstractLogCleanerIntegrationTest as an IntegrationTest




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] kafka pull request #3332: MINOR: Enable request and other debug loggers duri...

[ewencp]

GitHub user ewencp opened a pull request:

https://github.com/apache/kafka/pull/3332

MINOR: Enable request and other debug loggers during tests, directed at a 
NullAppender



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/ewencp/kafka minor-test-request-logger

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3332.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3332


commit 924e39aef5abbf64c53927b480dee9e62cc80bce
Author: Ewen Cheslack-Postava 
Date:   2017-06-14T06:22:01Z

MINOR: Enable request and other debug loggers during tests, directed at a 
NullAppender




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Comment Edited] (KAFKA-5444) Producer.send() will hang 8+ hours

[Hongyuan Li (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048709#comment-16048709
 ] 

Hongyuan Li edited comment on KAFKA-5444 at 6/14/17 6:15 AM:
-

[~huxi_2b]  
May be not duplicate of KAFKA-3552, the log only printed at kafka server.log 
not the client.  
the client only hang, which means stuck.
JVM version:{{oracle jdk 1.7.0_67 x64}}
Kafka version {{kafka_2.10-0.10.1.1}}
kafka server start cmd seen from jinfo lists below:
{code}
-Xmx1G -Xms1G -XX:+UseG1GC -XX:MaxGCPauseMillis=20 
-XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC 
-Djava.awt.headless=true 
-Xloggc:/root/kafka_2.10-0.10.1.1/bin/../logs/kafkaServer-gc.log -verbose:gc 
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps 
-Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.authenticate=false 
-Dcom.sun.management.jmxremote.ssl=false 
-Dkafka.logs.dir=/root/kafka_2.10-0.10.1.1/bin/../logs 
-Dlog4j.configuration=file:kafka_2.10-0.10.1.1/bin/../config/log4j.properties
{code}
you mean to remove {{-XX:+DisableExplicitGC}} ?


was (Author: hongyuan li):
[~huxi_2b]  
May be not duplicate of KAFKA-3552, the log only printed at kafka server.log 
not the client.  
the client only hang, which means stuck.
JVM version:{{oracle jdk 1.7.0_67 x64}}
kafka server start cmd seen from jinfo lists below:
{code}
-Xmx1G -Xms1G -XX:+UseG1GC -XX:MaxGCPauseMillis=20 
-XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC 
-Djava.awt.headless=true 
-Xloggc:/root/kafka_2.10-0.10.1.1/bin/../logs/kafkaServer-gc.log -verbose:gc 
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps 
-Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.authenticate=false 
-Dcom.sun.management.jmxremote.ssl=false 
-Dkafka.logs.dir=/root/kafka_2.10-0.10.1.1/bin/../logs 
-Dlog4j.configuration=file:kafka_2.10-0.10.1.1/bin/../config/log4j.properties
{code}
you mean to remove {{-XX:+DisableExplicitGC}} ?

> Producer.send() will hang 8+ hours
> --
>
> Key: KAFKA-5444
> URL: https://issues.apache.org/jira/browse/KAFKA-5444
> Project: Kafka
>  Issue Type: Bug
>Reporter: Hongyuan Li
>
> Frequent kafka old Producer open and close with cause the server hang with 
> lots of error messages logged in the server.log .In my occasion,we may 
> frequent open and close kafka producer,the procedure just like the code below:
> {code}
> Producer producer = ……
> producer.send(List lists);
> producer.close();
> {code}
> the error is below:
> {code}
> 2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
> (kafka.network.Processor)
> java.lang.OutOfMemoryError: Direct buffer memory
>   at java.nio.Bits.reserveMemory(Bits.java:658)
>   at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
>   at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
>   at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
>   at sun.nio.ch.IOUtil.read(IOUtil.java:195)
>   at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
>   at 
> org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
>   at kafka.network.Processor.poll(SocketServer.scala:476)
>   at kafka.network.Processor.run(SocketServer.scala:416)
>   at java.lang.Thread.run(Thread.java:745)
> {code}
> seen from all existing logs, all error is repeats of error above.
> Any good idea to solve this?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5245) KStream builder should capture serdes

[Evgeny Veretennikov (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5245?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048738#comment-16048738
 ] 

Evgeny Veretennikov commented on KAFKA-5245:


There are actually a lot of KStream methods, that require serdes again, not 
only groupByKey(). For example, there are print(), through(), to(), etc. We 
should pass serdes from builder.stream() to all such methods, am I right?

Also, behaviour of methods like groupByKey() will change after solving this 
ticket. Clients could be broken, in case they really need to use default 
serdes. Do we need KIP for this ticket?

> KStream builder should capture serdes 
> --
>
> Key: KAFKA-5245
> URL: https://issues.apache.org/jira/browse/KAFKA-5245
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.2.0, 0.10.2.1
>Reporter: Yeva Byzek
>Assignee: anugrah
>Priority: Minor
>  Labels: beginner, newbie
>
> Even if one specifies a serdes in `builder.stream`, later a call to 
> `groupByKey` may require the serdes again if it differs from the configured 
> streams app serdes. The preferred behavior is that if no serdes is provided 
> to `groupByKey`, it should use whatever was provided in `builder.stream` and 
> not what was in the app.
> From the current docs:
> “When to set explicit serdes: Variants of groupByKey exist to override the 
> configured default serdes of your application, which you must do if the key 
> and/or value types of the resulting KGroupedStream do not match the 
> configured default serdes.”



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Comment Edited] (KAFKA-5444) Producer.send() will hang 8+ hours

[Hongyuan Li (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048709#comment-16048709
 ] 

Hongyuan Li edited comment on KAFKA-5444 at 6/14/17 4:58 AM:
-

[~huxi_2b]  
May be not duplicate of KAFKA-3552, the log only printed at kafka server.log 
not the client.  
the client only hang, which means stuck.
JVM version:{{oracle jdk 1.7.0_67 x64}}
kafka server start cmd seen from jinfo lists below:
{code}
-Xmx1G -Xms1G -XX:+UseG1GC -XX:MaxGCPauseMillis=20 
-XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC 
-Djava.awt.headless=true 
-Xloggc:/root/kafka_2.10-0.10.1.1/bin/../logs/kafkaServer-gc.log -verbose:gc 
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps 
-Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.authenticate=false 
-Dcom.sun.management.jmxremote.ssl=false 
-Dkafka.logs.dir=/root/kafka_2.10-0.10.1.1/bin/../logs 
-Dlog4j.configuration=file:kafka_2.10-0.10.1.1/bin/../config/log4j.properties
{code}
you mean to remove {{-XX:+DisableExplicitGC}} ?


was (Author: hongyuan li):
[~huxi_2b]  
May be not duplicate of KAFKA-3552, the log only printed at kafka server.log 
not the client.  
the client only hang, which means stuck.
JVM version:oracle jdk 1.7.0_67 x64
kafka server start cmd seen from jinfo lists below:
{code}
-Xmx1G -Xms1G -XX:+UseG1GC -XX:MaxGCPauseMillis=20 
-XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC 
-Djava.awt.headless=true 
-Xloggc:/root/kafka_2.10-0.10.1.1/bin/../logs/kafkaServer-gc.log -verbose:gc 
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps 
-Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.authenticate=false 
-Dcom.sun.management.jmxremote.ssl=false 
-Dkafka.logs.dir=/root/kafka_2.10-0.10.1.1/bin/../logs 
-Dlog4j.configuration=file:kafka_2.10-0.10.1.1/bin/../config/log4j.properties
{code}
you mean to remove {{-XX:+DisableExplicitGC}} ?

> Producer.send() will hang 8+ hours
> --
>
> Key: KAFKA-5444
> URL: https://issues.apache.org/jira/browse/KAFKA-5444
> Project: Kafka
>  Issue Type: Bug
>Reporter: Hongyuan Li
>
> Frequent kafka old Producer open and close with cause the server hang with 
> lots of error messages logged in the server.log .In my occasion,we may 
> frequent open and close kafka producer,the procedure just like the code below:
> {code}
> Producer producer = ……
> producer.send(List lists);
> producer.close();
> {code}
> the error is below:
> {code}
> 2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
> (kafka.network.Processor)
> java.lang.OutOfMemoryError: Direct buffer memory
>   at java.nio.Bits.reserveMemory(Bits.java:658)
>   at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
>   at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
>   at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
>   at sun.nio.ch.IOUtil.read(IOUtil.java:195)
>   at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
>   at 
> org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
>   at kafka.network.Processor.poll(SocketServer.scala:476)
>   at kafka.network.Processor.run(SocketServer.scala:416)
>   at java.lang.Thread.run(Thread.java:745)
> {code}
> seen from all existing logs, all error is repeats of error above.
> Any good idea to solve this?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Comment Edited] (KAFKA-5444) Producer.send() will hang 8+ hours

[Hongyuan Li (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048709#comment-16048709
 ] 

Hongyuan Li edited comment on KAFKA-5444 at 6/14/17 4:57 AM:
-

[~huxi_2b]  
May be not duplicate of KAFKA-3552, the log only printed at kafka server.log 
not the client.  
the client only hang, which means stuck.
JVM version:oracle jdk 1.7.0_67 x64
kafka server start cmd seen from jinfo lists below:
{code}
-Xmx1G -Xms1G -XX:+UseG1GC -XX:MaxGCPauseMillis=20 
-XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC 
-Djava.awt.headless=true 
-Xloggc:/root/kafka_2.10-0.10.1.1/bin/../logs/kafkaServer-gc.log -verbose:gc 
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps 
-Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.authenticate=false 
-Dcom.sun.management.jmxremote.ssl=false 
-Dkafka.logs.dir=/root/kafka_2.10-0.10.1.1/bin/../logs 
-Dlog4j.configuration=file:kafka_2.10-0.10.1.1/bin/../config/log4j.properties
{code}
you mean to remove {{-XX:+DisableExplicitGC}} ?


was (Author: hongyuan li):










[~huxi_2b]  May be not duplicate of KAFKA-3552, the log only printed at kafka 
server.log not the client.  
the client only hang, which means stuck.
JVM version:oracle jdk 1.7.0_67 x64
kafka server start cmd seen from jinfo lists below:
{code}
-Xmx1G -Xms1G -XX:+UseG1GC -XX:MaxGCPauseMillis=20 
-XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC 
-Djava.awt.headless=true 
-Xloggc:/root/kafka_2.10-0.10.1.1/bin/../logs/kafkaServer-gc.log -verbose:gc 
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps 
-Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.authenticate=false 
-Dcom.sun.management.jmxremote.ssl=false 
-Dkafka.logs.dir=/root/kafka_2.10-0.10.1.1/bin/../logs 
-Dlog4j.configuration=file:kafka_2.10-0.10.1.1/bin/../config/log4j.properties
{code}
you mean to remove {{-XX:+DisableExplicitGC}} ?

> Producer.send() will hang 8+ hours
> --
>
> Key: KAFKA-5444
> URL: https://issues.apache.org/jira/browse/KAFKA-5444
> Project: Kafka
>  Issue Type: Bug
>Reporter: Hongyuan Li
>
> Frequent kafka old Producer open and close with cause the server hang with 
> lots of error messages logged in the server.log .In my occasion,we may 
> frequent open and close kafka producer,the procedure just like the code below:
> {code}
> Producer producer = ……
> producer.send(List lists);
> producer.close();
> {code}
> the error is below:
> {code}
> 2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
> (kafka.network.Processor)
> java.lang.OutOfMemoryError: Direct buffer memory
>   at java.nio.Bits.reserveMemory(Bits.java:658)
>   at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
>   at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
>   at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
>   at sun.nio.ch.IOUtil.read(IOUtil.java:195)
>   at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
>   at 
> org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
>   at kafka.network.Processor.poll(SocketServer.scala:476)
>   at kafka.network.Processor.run(SocketServer.scala:416)
>   at java.lang.Thread.run(Thread.java:745)
> {code}
> seen from all existing logs, all error is repeats of error above.
> Any good idea to solve this?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5444) Producer.send() will hang 8+ hours

[Hongyuan Li (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048709#comment-16048709
 ] 

Hongyuan Li commented on KAFKA-5444:












[~huxi_2b]  May be not duplicate of KAFKA-3552, the log only printed at kafka 
server.log not the client.  
the client only hang, which means stuck.
JVM version:oracle jdk 1.7.0_67 x64
kafka server start cmd seen from jinfo lists below:
{code}
-Xmx1G -Xms1G -XX:+UseG1GC -XX:MaxGCPauseMillis=20 
-XX:InitiatingHeapOccupancyPercent=35 -XX:+DisableExplicitGC 
-Djava.awt.headless=true 
-Xloggc:/root/kafka_2.10-0.10.1.1/bin/../logs/kafkaServer-gc.log -verbose:gc 
-XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+PrintGCTimeStamps 
-Dcom.sun.management.jmxremote 
-Dcom.sun.management.jmxremote.authenticate=false 
-Dcom.sun.management.jmxremote.ssl=false 
-Dkafka.logs.dir=/root/kafka_2.10-0.10.1.1/bin/../logs 
-Dlog4j.configuration=file:kafka_2.10-0.10.1.1/bin/../config/log4j.properties
{code}
you mean to remove {{-XX:+DisableExplicitGC}} ?

> Producer.send() will hang 8+ hours
> --
>
> Key: KAFKA-5444
> URL: https://issues.apache.org/jira/browse/KAFKA-5444
> Project: Kafka
>  Issue Type: Bug
>Reporter: Hongyuan Li
>
> Frequent kafka old Producer open and close with cause the server hang with 
> lots of error messages logged in the server.log .In my occasion,we may 
> frequent open and close kafka producer,the procedure just like the code below:
> {code}
> Producer producer = ……
> producer.send(List lists);
> producer.close();
> {code}
> the error is below:
> {code}
> 2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
> (kafka.network.Processor)
> java.lang.OutOfMemoryError: Direct buffer memory
>   at java.nio.Bits.reserveMemory(Bits.java:658)
>   at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
>   at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
>   at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
>   at sun.nio.ch.IOUtil.read(IOUtil.java:195)
>   at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
>   at 
> org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
>   at kafka.network.Processor.poll(SocketServer.scala:476)
>   at kafka.network.Processor.run(SocketServer.scala:416)
>   at java.lang.Thread.run(Thread.java:745)
> {code}
> seen from all existing logs, all error is repeats of error above.
> Any good idea to solve this?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5444) Producer.send() will hang 8+ hours

[Hongyuan Li (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Hongyuan Li updated KAFKA-5444:
---
Description: 
Frequent kafka old Producer open and close with cause the server hang with lots 
of error messages logged in the server.log .In my occasion,we may frequent open 
and close kafka producer,the procedure just like the code below:
{code}
Producer producer = ……
producer.send(List lists);
producer.close();
{code}
the error is below:
{code}
2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
(kafka.network.Processor)
java.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:658)
at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
at sun.nio.ch.IOUtil.read(IOUtil.java:195)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
at 
org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
at 
org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
at 
org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
at 
org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
at 
org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
at 
org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
at kafka.network.Processor.poll(SocketServer.scala:476)
at kafka.network.Processor.run(SocketServer.scala:416)
at java.lang.Thread.run(Thread.java:745)
{code}
seen from all existing logs, all error is repeats of error above.
Any good idea to solve this?

  was:
Frequent kafka old Producer open and close with cause the server hang with lots 
of error messages logged in the server.log .In my occasion,we may frequent open 
and close kafka producer,the procedure just like the code below:
{code}
Producer producer = ……
producer.send(List lists);
producer.close();
{code}
the error is below:
{code}
2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
(kafka.network.Processor)
java.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:658)
at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
at sun.nio.ch.IOUtil.read(IOUtil.java:195)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
at 
org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
at 
org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
at 
org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
at 
org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
at 
org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
at 
org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
at kafka.network.Processor.poll(SocketServer.scala:476)
at kafka.network.Processor.run(SocketServer.scala:416)
at java.lang.Thread.run(Thread.java:745)
{code}
the detailed error logs lists will be added in the attachment.
Any good idea to solve this?


> Producer.send() will hang 8+ hours
> --
>
> Key: KAFKA-5444
> URL: https://issues.apache.org/jira/browse/KAFKA-5444
> Project: Kafka
>  Issue Type: Bug
>Reporter: Hongyuan Li
>
> Frequent kafka old Producer open and close with cause the server hang with 
> lots of error messages logged in the server.log .In my occasion,we may 
> frequent open and close kafka producer,the procedure just like the code below:
> {code}
> Producer producer = ……
> producer.send(List lists);
> producer.close();
> {code}
> the error is below:
> {code}
> 2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
> (kafka.network.Processor)
> java.lang.OutOfMemoryError: Direct buffer memory
>   at java.nio.Bits.reserveMemory(Bits.java:658)
>   at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
>   at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
>   at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
>   at sun.nio.ch.IOUtil.read(IOUtil.java:195)
>   at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
>   at 
> org.apache.kafka.common.network.PlaintextTransportLayer.rea

Jenkins build is back to normal : kafka-trunk-jdk8 #1702

[Apache Jenkins Server]

See 

[jira] [Commented] (KAFKA-5443) Consumer should use last offset from batch to set next fetch offset

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5443?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048688#comment-16048688
 ] 

ASF GitHub Bot commented on KAFKA-5443:
---

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/3331

KAFKA-5443: Consumer should use last offset from batch to set next fetch 
offset



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka KAFKA-5443

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3331.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3331


commit b991c8185c5bce1cedb39dc5158afaed1b8db2f4
Author: Jason Gustafson 
Date:   2017-06-14T04:17:33Z

KAFKA-5443: Consumer should use last offset from batch to set next fetch 
offset




> Consumer should use last offset from batch to set next fetch offset
> ---
>
> Key: KAFKA-5443
> URL: https://issues.apache.org/jira/browse/KAFKA-5443
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
> Fix For: 0.11.0.0
>
>
> With message format v2, the log cleaner preserves the last offset in each 
> batch even if the last record is removed. Currently when the batch is 
> consumed by the consumer, we use the last record in the batch to determine 
> the next offset to fetch. So if the last record in the batch was removed 
> through compaction, the next fetch offset will still point to an offset in 
> the current batch and it will be refetched. In the worst case, if the fetch 
> size has room for that batch, the consumer will not be able to make progress. 
> To fix this, we should advance the next fetch offset to the last offset from 
> the batch once we have consumed that batch.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3331: KAFKA-5443: Consumer should use last offset from b...

[hachikuji]

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/3331

KAFKA-5443: Consumer should use last offset from batch to set next fetch 
offset



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka KAFKA-5443

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3331.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3331


commit b991c8185c5bce1cedb39dc5158afaed1b8db2f4
Author: Jason Gustafson 
Date:   2017-06-14T04:17:33Z

KAFKA-5443: Consumer should use last offset from batch to set next fetch 
offset




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-5444) Producer.send() will hang 8+ hours

[huxihx (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048674#comment-16048674
 ] 

huxihx commented on KAFKA-5444:
---

Is it a duplicate of 
[KAFKA-3552|https://issues.apache.org/jira/browse/KAFKA-3552]?  What's your 
Kafka version and JVM version?  And since `reserveMemory` triggers a System.gc 
if no extra space is found for allocating direct byte buffer, check your env to 
see if System.gc is explicitly disabled.

> Producer.send() will hang 8+ hours
> --
>
> Key: KAFKA-5444
> URL: https://issues.apache.org/jira/browse/KAFKA-5444
> Project: Kafka
>  Issue Type: Bug
>Reporter: Hongyuan Li
>
> Frequent kafka old Producer open and close with cause the server hang with 
> lots of error messages logged in the server.log .In my occasion,we may 
> frequent open and close kafka producer,the procedure just like the code below:
> {code}
> Producer producer = ……
> producer.send(List lists);
> producer.close();
> {code}
> the error is below:
> {code}
> 2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
> (kafka.network.Processor)
> java.lang.OutOfMemoryError: Direct buffer memory
>   at java.nio.Bits.reserveMemory(Bits.java:658)
>   at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
>   at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
>   at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
>   at sun.nio.ch.IOUtil.read(IOUtil.java:195)
>   at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
>   at 
> org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
>   at kafka.network.Processor.poll(SocketServer.scala:476)
>   at kafka.network.Processor.run(SocketServer.scala:416)
>   at java.lang.Thread.run(Thread.java:745)
> {code}
> the detailed error logs lists will be added in the attachment.
> Any good idea to solve this?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3330: MINOR: Javadoc for ExtendedSerializer and Extended...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3330


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-5444) Producer.send() will hang 8+ hours

[Hongyuan Li (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5444?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Hongyuan Li updated KAFKA-5444:
---
Summary: Producer.send() will hang 8+ hours  (was: Producer.send() will 
hang 8 hours, )

> Producer.send() will hang 8+ hours
> --
>
> Key: KAFKA-5444
> URL: https://issues.apache.org/jira/browse/KAFKA-5444
> Project: Kafka
>  Issue Type: Bug
>Reporter: Hongyuan Li
>
> Frequent kafka old Producer open and close with cause the server hang with 
> lots of error messages logged in the server.log .In my occasion,we may 
> frequent open and close kafka producer,the procedure just like the code below:
> {code}
> Producer producer = ……
> producer.send(List lists);
> producer.close();
> {code}
> the error is below:
> {code}
> 2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
> (kafka.network.Processor)
> java.lang.OutOfMemoryError: Direct buffer memory
>   at java.nio.Bits.reserveMemory(Bits.java:658)
>   at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
>   at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
>   at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
>   at sun.nio.ch.IOUtil.read(IOUtil.java:195)
>   at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
>   at 
> org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
>   at kafka.network.Processor.poll(SocketServer.scala:476)
>   at kafka.network.Processor.run(SocketServer.scala:416)
>   at java.lang.Thread.run(Thread.java:745)
> {code}
> the detailed error logs lists will be added in the attachment.
> Any good idea to solve this?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (KAFKA-5444) Producer.send() will hang 8 hours,

[Hongyuan Li (JIRA)]

Hongyuan Li created KAFKA-5444:
--

 Summary: Producer.send() will hang 8 hours, 
 Key: KAFKA-5444
 URL: https://issues.apache.org/jira/browse/KAFKA-5444
 Project: Kafka
  Issue Type: Bug
Reporter: Hongyuan Li


Frequent kafka old Producer open and close with cause the server hang with lots 
of error messages logged in the server.log .In my occasion,we may frequent open 
and close kafka producer,the procedure just like the code below:
{code}
Producer producer = ……
producer.send(List lists);
producer.close();
{code}
the error is below:
{code}
2017-06-13 00:00:00,084] ERROR Processor got uncaught exception. 
(kafka.network.Processor)
java.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:658)
at java.nio.DirectByteBuffer.(DirectByteBuffer.java:123)
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:174)
at sun.nio.ch.IOUtil.read(IOUtil.java:195)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
at 
org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
at 
org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
at 
org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
at 
org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
at 
org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
at 
org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:343)
at org.apache.kafka.common.network.Selector.poll(Selector.java:291)
at kafka.network.Processor.poll(SocketServer.scala:476)
at kafka.network.Processor.run(SocketServer.scala:416)
at java.lang.Thread.run(Thread.java:745)
{code}
the detailed error logs lists will be added in the attachment.
Any good idea to solve this?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-3455) Connect custom processors with the streams DSL

[Bobby Calderwood (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-3455?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048626#comment-16048626
 ] 

Bobby Calderwood commented on KAFKA-3455:
-

Hi Michal Borowiecki,

Sorry for the late reply.  I am trying to implement both {{Transformer}} and 
{{Processor}} on the same class.  For some interesting use-cases in a Clojure 
compatibility library for Kafka Streams that I'm writing, I'd like to hook a 
single piece of logic into both the high-level (via {{Transformer}}) and 
low-level (via {{Processor}}) APIs.  However, when implementing both 
interfaces, I encounter the following error due to differing signatures of the 
respective {{punctuate(long)}} methods:

{code:none}
/TransducerProcessor.java
Error:Error:line (44)java: method punctuate(long) is already defined in 
class kafka_streams_clojure.TransducerProcessor
Error:Error:line (10)java: kafka_streams_clojure.TransducerProcessor is not 
abstract and does not override abstract method punctuate(long) in 
org.apache.kafka.streams.kstream.Transformer
Error:Error:line (40)java: punctuate(long) in 
kafka_streams_clojure.TransducerProcessor cannot implement punctuate(long) in 
org.apache.kafka.streams.kstream.Transformer
  return type void is not compatible with java.lang.Object
{code}

I believe you're right about AutoCloseable, it's been a while since I've been 
in Java-land :-)

Yes, I believe that once the incompatible punctuate methods are deprecated and 
then removed, my issue would be resolved.

> Connect custom processors with the streams DSL
> --
>
> Key: KAFKA-3455
> URL: https://issues.apache.org/jira/browse/KAFKA-3455
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Affects Versions: 0.10.0.1
>Reporter: Jonathan Bender
>  Labels: user-experience
>
> From the kafka users email thread, we discussed the idea of connecting custom 
> processors with topologies defined from the Streams DSL (and being able to 
> sink data from the processor).  Possibly this could involve exposing the 
> underlying processor's name in the streams DSL so it can be connected with 
> the standard processor API.
> {quote}
> Thanks for the feedback. This is definitely something we wanted to support
> in the Streams DSL.
> One tricky thing, though, is that some operations do not translate to a
> single processor, but a sub-graph of processors (think of a stream-stream
> join, which is translated to actually 5 processors for windowing / state
> queries / merging, each with a different internal name). So how to define
> the API to return the processor name needs some more thinking.
> {quote}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5432) producer and consumer SocketTimeoutException

[huxihx (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5432?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048615#comment-16048615
 ] 

huxihx commented on KAFKA-5432:
---

I might not see rolling new log segment as an indicator of the cause.  Is that 
possible that those CLOSE_WAITs lead to the SocketTimeoutException thrown both 
by consumers and producers?

> producer and consumer  SocketTimeoutException
> -
>
> Key: KAFKA-5432
> URL: https://issues.apache.org/jira/browse/KAFKA-5432
> Project: Kafka
>  Issue Type: Bug
>  Components: network
>Affects Versions: 0.10.2.0
> Environment: os:Red Hat 4.4.7-17
> java:
> java version "1.8.0_131"
> Java(TM) SE Runtime Environment (build 1.8.0_131-b11)
> Java HotSpot(TM) 64-Bit Server VM (build 25.131-b11, mixed mode)
>Reporter: Jian Lin
> Attachments: server.properties
>
>
> Hey all, I met a strange problem, hope someone can help me.
> The program ran normally for a week, and I did not do any changes, but today 
> it reported a mistake suddenly
> Producer error log:
> {code:java}
> 2017-06-12 10:46:01[qtp958382397-80:591423838]-[WARN] Failed to send producer 
> request with correlation id 234645 to broker 176 with data for partitions 
> [sms,3]
> java.net.SocketTimeoutException
>   at 
> sun.nio.ch.SocketAdaptor$SocketInputStream.read(SocketAdaptor.java:229)
>   at sun.nio.ch.ChannelInputStream.read(ChannelInputStream.java:103)
>   at 
> java.nio.channels.Channels$ReadableByteChannelImpl.read(Channels.java:385)
>   at kafka.utils.Utils$.read(Utils.scala:380)
>   at 
> kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
>   at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
>   at 
> kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
>   at kafka.network.BlockingChannel.receive(BlockingChannel.scala:111)
>   at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
>   at 
> kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
>   at 
> kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:103)
>   at 
> kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
>   at 
> kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:103)
>   at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>   at 
> kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:102)
>   at 
> kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
>   at 
> kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:102)
>   at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>   at kafka.producer.SyncProducer.send(SyncProducer.scala:101)
>   at 
> kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$send(DefaultEventHandler.scala:255)
>   at 
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:106)
>   at 
> kafka.producer.async.DefaultEventHandler$$anonfun$dispatchSerializedData$2.apply(DefaultEventHandler.scala:100)
>   at 
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:95)
>   at 
> scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:95)
>   at scala.collection.Iterator$class.foreach(Iterator.scala:772)
>   at 
> scala.collection.mutable.HashTable$$anon$1.foreach(HashTable.scala:157)
>   at 
> scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:190)
>   at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:45)
>   at scala.collection.mutable.HashMap.foreach(HashMap.scala:95)
>   at 
> kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:100)
>   at 
> kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:72)
>   at kafka.producer.Producer.send(Producer.scala:77)
>   at kafka.javaapi.producer.Producer.send(Producer.scala:33)
> {code}
> Consumer error log:
> {code:java}
> 2017-06-12 
> 10:46:52[ConsumerFetcherThread-sms-consumer-group1_zw_78_64-1496632739724-69516149-0-176:602874523]-[INFO]
>  Reconnect due to socket error: java.net.SocketTimeoutException
> 2017-06-12 
> 10:47:22[ConsumerFetcherThread-sms-consumer-group1_zw_78_64-1496632739724-69516149-0-176:602904544]-[WARN]
>  
> [ConsumerFetcherThread-sms-consumer-group1_zw_78_64-1496632739724-69516149-0-176],
>  Error in fetch Name: FetchRequest; Version: 0; CorrelationId: 6060231; 
> ClientId: sms-consumer-group1; ReplicaId: -1; MaxWait: 100 ms; MinBytes: 1 
> bytes; RequestInfo: [sms,0] -> Par

Build failed in Jenkins: kafka-trunk-jdk8 #1701

[Apache Jenkins Server]

See 


Changes:

[jason] KAFKA-5438; Fix UnsupportedOperationException in WriteTxnMarkersRequest

[me] MINOR: Add unit tests for PluginDesc in Connect.

--
[...truncated 4.21 MB...]

kafka.controller.ControllerIntegrationTest > 
testControllerEpochPersistsWhenAllBrokersDown PASSED

kafka.controller.ControllerIntegrationTest > 
testTopicCreationWithOfflineReplica STARTED

kafka.controller.ControllerIntegrationTest > 
testTopicCreationWithOfflineReplica PASSED

kafka.controller.ControllerIntegrationTest > 
testPartitionReassignmentResumesAfterReplicaComesOnline STARTED

kafka.controller.ControllerIntegrationTest > 
testPartitionReassignmentResumesAfterReplicaComesOnline PASSED

kafka.controller.ControllerIntegrationTest > 
testLeaderAndIsrWhenEntireIsrOfflineAndUncleanLeaderElectionDisabled STARTED

kafka.controller.ControllerIntegrationTest > 
testLeaderAndIsrWhenEntireIsrOfflineAndUncleanLeaderElectionDisabled PASSED

kafka.controller.ControllerIntegrationTest > 
testTopicPartitionExpansionWithOfflineReplica STARTED

kafka.controller.ControllerIntegrationTest > 
testTopicPartitionExpansionWithOfflineReplica PASSED

kafka.controller.ControllerIntegrationTest > 
testPreferredReplicaLeaderElectionWithOfflinePreferredReplica STARTED

kafka.controller.ControllerIntegrationTest > 
testPreferredReplicaLeaderElectionWithOfflinePreferredReplica PASSED

kafka.controller.ControllerIntegrationTest > 
testAutoPreferredReplicaLeaderElection STARTED

kafka.controller.ControllerIntegrationTest > 
testAutoPreferredReplicaLeaderElection PASSED

kafka.controller.ControllerIntegrationTest > testTopicCreation STARTED

kafka.controller.ControllerIntegrationTest > testTopicCreation PASSED

kafka.controller.ControllerIntegrationTest > testPartitionReassignment STARTED

kafka.controller.ControllerIntegrationTest > testPartitionReassignment PASSED

kafka.controller.ControllerIntegrationTest > testTopicPartitionExpansion STARTED

kafka.controller.ControllerIntegrationTest > testTopicPartitionExpansion PASSED

kafka.controller.ControllerIntegrationTest > 
testControllerMoveIncrementsControllerEpoch STARTED

kafka.controller.ControllerIntegrationTest > 
testControllerMoveIncrementsControllerEpoch PASSED

kafka.controller.ControllerIntegrationTest > 
testLeaderAndIsrWhenEntireIsrOfflineAndUncleanLeaderElectionEnabled STARTED

kafka.controller.ControllerIntegrationTest > 
testLeaderAndIsrWhenEntireIsrOfflineAndUncleanLeaderElectionEnabled PASSED

kafka.controller.ControllerIntegrationTest > testEmptyCluster STARTED

kafka.controller.ControllerIntegrationTest > testEmptyCluster PASSED

kafka.controller.ControllerIntegrationTest > testPreferredReplicaLeaderElection 
STARTED

kafka.controller.ControllerIntegrationTest > testPreferredReplicaLeaderElection 
PASSED

kafka.controller.ControllerEventManagerTest > testEventThatThrowsException 
STARTED

kafka.controller.ControllerEventManagerTest > testEventThatThrowsException 
PASSED

kafka.controller.ControllerEventManagerTest > testSuccessfulEvent STARTED

kafka.controller.ControllerEventManagerTest > testSuccessfulEvent PASSED

kafka.controller.ControllerFailoverTest > testHandleIllegalStateException 
STARTED

kafka.controller.ControllerFailoverTest > testHandleIllegalStateException PASSED

kafka.network.SocketServerTest > testGracefulClose STARTED

kafka.network.SocketServerTest > testGracefulClose PASSED

kafka.network.SocketServerTest > testClientDisconnectionUpdatesRequestMetrics 
STARTED

kafka.network.SocketServerTest > testClientDisconnectionUpdatesRequestMetrics 
PASSED

kafka.network.SocketServerTest > testProcessorMetricsTags STARTED

kafka.network.SocketServerTest > testProcessorMetricsTags PASSED

kafka.network.SocketServerTest > testMaxConnectionsPerIp STARTED

kafka.network.SocketServerTest > testMaxConnectionsPerIp PASSED

kafka.network.SocketServerTest > 
testBrokerSendAfterChannelClosedUpdatesRequestMetrics STARTED

kafka.network.SocketServerTest > 
testBrokerSendAfterChannelClosedUpdatesRequestMetrics PASSED

kafka.network.SocketServerTest > simpleRequest STARTED

kafka.network.SocketServerTest > simpleRequest PASSED

kafka.network.SocketServerTest > testMetricCollectionAfterShutdown STARTED

kafka.network.SocketServerTest > testMetricCollectionAfterShutdown PASSED

kafka.network.SocketServerTest > testSessionPrincipal STARTED

kafka.network.SocketServerTest > testSessionPrincipal PASSED

kafka.network.SocketServerTest > testMaxConnectionsPerIpOverrides STARTED

kafka.network.SocketServerTest > testMaxConnectionsPerIpOverrides PASSED

kafka.network.SocketServerTest > testSocketsCloseOnShutdown STARTED

kafka.network.SocketServerTest > testSocketsCloseOnShutdown PASSED

kafka.network.SocketServerTest > testSslSocketServer STARTED

kafka.network.SocketServerTest > testSslSocketServer PASSED

kafka.network.SocketServerTest > tooBigRequestIsRejected STARTED

[jira] [Commented] (KAFKA-5007) Kafka Replica Fetcher Thread- Resource Leak

[huxihx (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5007?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048605#comment-16048605
 ] 

huxihx commented on KAFKA-5007:
---

[~joseph.alias...@gmail.com] org.apache.kafka.common.network.Selector

> Kafka Replica Fetcher Thread- Resource Leak
> ---
>
> Key: KAFKA-5007
> URL: https://issues.apache.org/jira/browse/KAFKA-5007
> Project: Kafka
>  Issue Type: Bug
>  Components: core, network
>Affects Versions: 0.10.0.0, 0.10.1.1, 0.10.2.0
> Environment: Centos 7
> Jave 8
>Reporter: Joseph Aliase
>Priority: Critical
>  Labels: reliability
> Attachments: jstack-kafka.out, jstack-zoo.out, lsofkafka.txt, 
> lsofzookeeper.txt
>
>
> Kafka is running out of open file descriptor when system network interface is 
> done.
> Issue description:
> We have a Kafka Cluster of 5 node running on version 0.10.1.1. The open file 
> descriptor for the account running Kafka is set to 10.
> During an upgrade, network interface went down. Outage continued for 12 hours 
> eventually all the broker crashed with java.io.IOException: Too many open 
> files error.
> We repeated the test in a lower environment and observed that Open Socket 
> count keeps on increasing while the NIC is down.
> We have around 13 topics with max partition size of 120 and number of replica 
> fetcher thread is set to 8.
> Using an internal monitoring tool we observed that Open Socket descriptor   
> for the broker pid continued to increase although NIC was down leading to  
> Open File descriptor error. 



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5007) Kafka Replica Fetcher Thread- Resource Leak

[Joseph Aliase (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5007?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048586#comment-16048586
 ] 

Joseph Aliase commented on KAFKA-5007:
--

[~huxi_2b] Please let me know the class this block belongs too

> Kafka Replica Fetcher Thread- Resource Leak
> ---
>
> Key: KAFKA-5007
> URL: https://issues.apache.org/jira/browse/KAFKA-5007
> Project: Kafka
>  Issue Type: Bug
>  Components: core, network
>Affects Versions: 0.10.0.0, 0.10.1.1, 0.10.2.0
> Environment: Centos 7
> Jave 8
>Reporter: Joseph Aliase
>Priority: Critical
>  Labels: reliability
> Attachments: jstack-kafka.out, jstack-zoo.out, lsofkafka.txt, 
> lsofzookeeper.txt
>
>
> Kafka is running out of open file descriptor when system network interface is 
> done.
> Issue description:
> We have a Kafka Cluster of 5 node running on version 0.10.1.1. The open file 
> descriptor for the account running Kafka is set to 10.
> During an upgrade, network interface went down. Outage continued for 12 hours 
> eventually all the broker crashed with java.io.IOException: Too many open 
> files error.
> We repeated the test in a lower environment and observed that Open Socket 
> count keeps on increasing while the NIC is down.
> We have around 13 topics with max partition size of 120 and number of replica 
> fetcher thread is set to 8.
> Using an internal monitoring tool we observed that Open Socket descriptor   
> for the broker pid continued to increase although NIC was down leading to  
> Open File descriptor error. 



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3319: MINOR: Verify mocks in all WorkerTest tests and do...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3319


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] kafka pull request #3330: MINOR: Javadoc for ExtendedSerializer and Extended...

[ijuma]

GitHub user ijuma opened a pull request:

https://github.com/apache/kafka/pull/3330

MINOR: Javadoc for ExtendedSerializer and ExtendedDeserializer

And add warning about usage.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/ijuma/kafka extended-serializer-javadoc

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3330.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3330


commit f90fa2579ca78826ea082ebf9fb6c9f80f478e8f
Author: Ismael Juma 
Date:   2017-06-13T14:09:20Z

Javadoc for ExtendedSerializer and ExtendedDeserializer and add warning 
about usage




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Resolved] (KAFKA-5406) NoNodeException result in rebalance failed

[xiaoguy (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5406?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

xiaoguy resolved KAFKA-5406.

Resolution: Done

it's the zk problem, the project used low version zk ,and it's can't work as 
normal when the network has problem, that is it 

tks all guys 

> NoNodeException result in rebalance failed
> --
>
> Key: KAFKA-5406
> URL: https://issues.apache.org/jira/browse/KAFKA-5406
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.8.2.2, 0.10.0.0
> Environment: windows8.1 centos6.4
>Reporter: xiaoguy
>Priority: Critical
>  Labels: easyfix, patch
> Attachments: log.log
>
>
> hey guys , I got this problem this days,
> because of the network is unstable， consumer rebalance failed after 5 times 
> ,the log shows that zk path /consumers/$(groupIdName)/ids/ is empty,
> consumer seems can't register after network recovered, so i got the kafka 
> source code (0.8.2.2) and found the 
> consumer/ZookeeperConsumerConnector$ZKSessionExpireListener handleNewSession 
> won't call , and handleStateChanged do nothing,
> so i change the code like this ,and it seems works ,  and i checked 0.10.0.0 
> version, the same problem, is this a bug ? i'm confused , thank you.
>   def handleStateChanged(state: KeeperState) {
>// do nothing, since zkclient will do reconnect for us.
>  if(state==KeeperState.SyncConnected){
>   handleNewSession()
>  }
>   System.err.println("ZKSessionExpireListener 
> handleStateChanged-state:"+state+""+state.getIntValue)
>   }



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5435) Produce state lost if no snapshot retained

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5435?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5435:
---
Status: Patch Available  (was: Open)

> Produce state lost if no snapshot retained
> --
>
> Key: KAFKA-5435
> URL: https://issues.apache.org/jira/browse/KAFKA-5435
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>Priority: Blocker
> Fix For: 0.11.0.0
>
>
> We have an optimization in {{Log}} to avoid the need to scan the log to build 
> producer state during the upgrade path. Basically, if no producer snapshot 
> exists, then we assume that it's an upgrade and take a new snapshot from the 
> end of the log. Unfortunately, it can happen that snapshot files are never 
> created or are deleted through truncation. Upon reinitialization, this can 
> cause the optimization above to kick in and we lose the current state of all 
> producers.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5214) Re-add KafkaAdminClient#apiVersions

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5214?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5214:
---
Description: We removed KafkaAdminClient#apiVersions just before 0.11.0.0 
to give us a bit more time to iterate on it before it's included in a release. 
We should add the relevant methods back.  (was: KafkaAdminClient#apiVersions 
should not refer to internal classes like ApiKeys, NodeApiVersions, etc.  
Instead, we should have stable public classes to represent these things in the 
API.)

> Re-add KafkaAdminClient#apiVersions
> ---
>
> Key: KAFKA-5214
> URL: https://issues.apache.org/jira/browse/KAFKA-5214
> Project: Kafka
>  Issue Type: Bug
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
>Priority: Blocker
> Fix For: 0.11.1.0
>
>
> We removed KafkaAdminClient#apiVersions just before 0.11.0.0 to give us a bit 
> more time to iterate on it before it's included in a release. We should add 
> the relevant methods back.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5214) Re-add KafkaAdminClient#apiVersions

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5214?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5214:
---
Summary: Re-add KafkaAdminClient#apiVersions  (was: 
KafkaAdminClient#apiVersions should return a public class)

> Re-add KafkaAdminClient#apiVersions
> ---
>
> Key: KAFKA-5214
> URL: https://issues.apache.org/jira/browse/KAFKA-5214
> Project: Kafka
>  Issue Type: Bug
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
>Priority: Blocker
> Fix For: 0.11.1.0
>
>
> KafkaAdminClient#apiVersions should not refer to internal classes like 
> ApiKeys, NodeApiVersions, etc.  Instead, we should have stable public classes 
> to represent these things in the API.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3324: Temporarily remove the apiVersions API for 0.11

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3324


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-5214) KafkaAdminClient#apiVersions should return a public class

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5214?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5214:
---
Fix Version/s: (was: 0.11.0.0)
   0.11.1.0

> KafkaAdminClient#apiVersions should return a public class
> -
>
> Key: KAFKA-5214
> URL: https://issues.apache.org/jira/browse/KAFKA-5214
> Project: Kafka
>  Issue Type: Bug
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
>Priority: Blocker
> Fix For: 0.11.1.0
>
>
> KafkaAdminClient#apiVersions should not refer to internal classes like 
> ApiKeys, NodeApiVersions, etc.  Instead, we should have stable public classes 
> to represent these things in the API.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (KAFKA-5443) Consumer should use last offset from batch to set next fetch offset

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-5443:
--

 Summary: Consumer should use last offset from batch to set next 
fetch offset
 Key: KAFKA-5443
 URL: https://issues.apache.org/jira/browse/KAFKA-5443
 Project: Kafka
  Issue Type: Sub-task
Reporter: Jason Gustafson
Assignee: Jason Gustafson
 Fix For: 0.11.0.0


With message format v2, the log cleaner preserves the last offset in each batch 
even if the last record is removed. Currently when the batch is consumed by the 
consumer, we use the last record in the batch to determine the next offset to 
fetch. So if the last record in the batch was removed through compaction, the 
next fetch offset will still point to an offset in the current batch and it 
will be refetched. In the worst case, if the fetch size has room for that 
batch, the consumer will not be able to make progress. To fix this, we should 
advance the next fetch offset to the last offset from the batch once we have 
consumed that batch.




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5413) Log cleaner fails due to large offset in segment file

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5413?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5413:
---
Fix Version/s: 0.11.0.1

> Log cleaner fails due to large offset in segment file
> -
>
> Key: KAFKA-5413
> URL: https://issues.apache.org/jira/browse/KAFKA-5413
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.10.2.0, 0.10.2.1
> Environment: Ubuntu 14.04 LTS, Oracle Java 8u92, kafka_2.11-0.10.2.0
>Reporter: Nicholas Ngorok
>  Labels: reliability
> Fix For: 0.11.0.1
>
> Attachments: .index.cleaned, 
> .log, .log.cleaned, 
> .timeindex.cleaned, 002147422683.log, 
> kafka-5413.patch
>
>
> The log cleaner thread in our brokers is failing with the trace below
> {noformat}
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 0 in log __consumer_offsets-12 (largest timestamp Thu Jun 08 
> 15:48:59 PDT 2017) into 0, retaining deletes. (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 2147343575 in log __consumer_offsets-12 (largest timestamp 
> Thu Jun 08 15:49:06 PDT 2017) into 0, retaining deletes. 
> (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,834] ERROR {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Error due to  (kafka.log.LogCleaner)
> java.lang.IllegalArgumentException: requirement failed: largest offset in 
> message set can not be safely converted to relative offset.
> at scala.Predef$.require(Predef.scala:224)
> at kafka.log.LogSegment.append(LogSegment.scala:109)
> at kafka.log.Cleaner.cleanInto(LogCleaner.scala:478)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:405)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:401)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:401)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:363)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:362)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.clean(LogCleaner.scala:362)
> at 
> kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:241)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:220)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
> [2017-06-08 15:49:54,835] INFO {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Stopped  (kafka.log.LogCleaner)
> {noformat}
> This seems to point at the specific line [here| 
> https://github.com/apache/kafka/blob/0.11.0/core/src/main/scala/kafka/log/LogSegment.scala#L92]
>  in the kafka src where the difference is actually larger than MAXINT as both 
> baseOffset and offset are of type long. It was introduced in this [pr| 
> https://github.com/apache/kafka/pull/2210/files/56d1f8196b77a47b176b7bbd1e4220a3be827631]
> These were the outputs of dumping the first two log segments
> {noformat}
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0.log
> Dumping /kafka-logs/__consumer_offsets-12/.log
> Starting offset: 0
> offset: 1810054758 position: 0 NoTimestampType: -1 isvalid: true payloadsize: 
> -1 magic: 0 compresscodec: NONE crc: 3127861909 keysize: 34
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0002147343575.log
> Dumping /kafka-logs/__consumer_offsets-12/002147343575.log
> Starting offset: 2147343575
> offset: 2147539884 position: 0 NoTimestampType: -1 isvalid: true paylo
> adsize: -1 magic: 0 compresscodec: NONE crc: 2282192097 keysize: 34
> {noformat}
> My guess is that since 2147539884 is larger than MAXINT, we are hitting this 
> exception. Was there a specific reason, this check was added in 0.10.2?
> E.g. if the first offset is a key = "key 0" and then we have MAXINT + 1 of 
> "key 1" following, wouldn't we run into this situation whenever the log 
> cleaner runs?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3309: MINOR: NetworkClient#disconnect should not erase c...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3309


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-764) Race Condition in Broker Registration after ZooKeeper disconnect

[Robert P. Thille (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-764?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Robert P. Thille updated KAFKA-764:
---
Attachment: BPPF_2900-Broker_Logs.tbz2

I believe the issues started somewhere around the time of these log messages:
[2017-05-25 07:08:25,528] INFO [Controller 2]: Broker 2 resigned as the 
controller (kafka.controller.KafkaController)
[2017-05-25 07:09:02,522] INFO [Controller 2]: Broker 2 resigned as the 
controller (kafka.controller.KafkaController)


> Race Condition in Broker Registration after ZooKeeper disconnect
> 
>
> Key: KAFKA-764
> URL: https://issues.apache.org/jira/browse/KAFKA-764
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.7.1
>Reporter: Bob Cotton
> Attachments: BPPF_2900-Broker_Logs.tbz2
>
>
> When running our ZooKeepers in VMware, occasionally all the keepers 
> simultaneously pause long enough for the Kafka clients to time out and then 
> the keepers simultaneously un-pause.
> When this happens, the zk clients disconnect from ZooKeeper. When ZooKeeper 
> comes back ZkUtils.createEphemeralPathExpectConflict discovers the node id of 
> itself and does not re-register the broker id node and the function call 
> succeeds. Then ZooKeeper figures out the broker disconnected from the keeper 
> and deletes the ephemeral node *after* allowing the consumer to read the data 
> in the /brokers/ids/x node.  The broker then goes on to register all the 
> topics, etc.  When consumers connect, they see topic nodes associated with 
> the broker but thy can't find the broker node to get connection information 
> for the broker, sending them into a rebalance loop until they reach 
> rebalance.retries.max and fail.
> This might also be a ZooKeeper issue, but the desired behavior for a 
> disconnect case might be, if the broker node is found to explicitly delete 
> and recreate it.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-764) Race Condition in Broker Registration after ZooKeeper disconnect

[Robert P. Thille (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-764?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048552#comment-16048552
 ] 

Robert P. Thille commented on KAFKA-764:


I believe we saw this issue, or something very similar.  
During a load test, we had a 3-node Kafka cluster which got into a confused 
state: 
Brokers 0 and 1 were happy and were listed in /brokers/ids/X in ZK, and Broker 
2 was connected to ZK, but not listed in /brokers/ids/2 and brokers 0 & 1 had 
no connections to broker 2. 
Broker 2 was happily accepting new messages produced to it for hours.  
Eventually, it did rejoin the cluster, but the published messages were lost as 
the 0 & 1 brokers seemingly outvoted broker 2 about the partitions.

> Race Condition in Broker Registration after ZooKeeper disconnect
> 
>
> Key: KAFKA-764
> URL: https://issues.apache.org/jira/browse/KAFKA-764
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.7.1
>Reporter: Bob Cotton
>
> When running our ZooKeepers in VMware, occasionally all the keepers 
> simultaneously pause long enough for the Kafka clients to time out and then 
> the keepers simultaneously un-pause.
> When this happens, the zk clients disconnect from ZooKeeper. When ZooKeeper 
> comes back ZkUtils.createEphemeralPathExpectConflict discovers the node id of 
> itself and does not re-register the broker id node and the function call 
> succeeds. Then ZooKeeper figures out the broker disconnected from the keeper 
> and deletes the ephemeral node *after* allowing the consumer to read the data 
> in the /brokers/ids/x node.  The broker then goes on to register all the 
> topics, etc.  When consumers connect, they see topic nodes associated with 
> the broker but thy can't find the broker node to get connection information 
> for the broker, sending them into a rebalance loop until they reach 
> rebalance.retries.max and fail.
> This might also be a ZooKeeper issue, but the desired behavior for a 
> disconnect case might be, if the broker node is found to explicitly delete 
> and recreate it.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Assigned] (KAFKA-5020) Update protocol documentation to mention message format v2

[Apurva Mehta (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5020?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Apurva Mehta reassigned KAFKA-5020:
---

Assignee: Apurva Mehta

> Update protocol documentation to mention message format v2
> --
>
> Key: KAFKA-5020
> URL: https://issues.apache.org/jira/browse/KAFKA-5020
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Ismael Juma
>Assignee: Apurva Mehta
>Priority: Critical
>  Labels: documentation, exactly-once
> Fix For: 0.11.0.0
>
>
> Sections 5.3, 5.4 and 5.5 should be updated:
> https://kafka.apache.org/documentation/#messages
> We may want to mention record batches along with message sets here:
> https://kafka.apache.org/protocol#protocol_message_sets
> And we should update the wiki page linked from the protocol documentation:
> https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-Messagesets



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5413) Log cleaner fails due to large offset in segment file

[Kelvin Rutt (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5413?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Kelvin Rutt updated KAFKA-5413:
---
Attachment: kafka-5413.patch

This is the patch I did.  It's basic but does the job.

> Log cleaner fails due to large offset in segment file
> -
>
> Key: KAFKA-5413
> URL: https://issues.apache.org/jira/browse/KAFKA-5413
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.10.2.0, 0.10.2.1
> Environment: Ubuntu 14.04 LTS, Oracle Java 8u92, kafka_2.11-0.10.2.0
>Reporter: Nicholas Ngorok
>  Labels: reliability
> Attachments: .index.cleaned, 
> .log, .log.cleaned, 
> .timeindex.cleaned, 002147422683.log, 
> kafka-5413.patch
>
>
> The log cleaner thread in our brokers is failing with the trace below
> {noformat}
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 0 in log __consumer_offsets-12 (largest timestamp Thu Jun 08 
> 15:48:59 PDT 2017) into 0, retaining deletes. (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 2147343575 in log __consumer_offsets-12 (largest timestamp 
> Thu Jun 08 15:49:06 PDT 2017) into 0, retaining deletes. 
> (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,834] ERROR {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Error due to  (kafka.log.LogCleaner)
> java.lang.IllegalArgumentException: requirement failed: largest offset in 
> message set can not be safely converted to relative offset.
> at scala.Predef$.require(Predef.scala:224)
> at kafka.log.LogSegment.append(LogSegment.scala:109)
> at kafka.log.Cleaner.cleanInto(LogCleaner.scala:478)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:405)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:401)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:401)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:363)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:362)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.clean(LogCleaner.scala:362)
> at 
> kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:241)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:220)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
> [2017-06-08 15:49:54,835] INFO {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Stopped  (kafka.log.LogCleaner)
> {noformat}
> This seems to point at the specific line [here| 
> https://github.com/apache/kafka/blob/0.11.0/core/src/main/scala/kafka/log/LogSegment.scala#L92]
>  in the kafka src where the difference is actually larger than MAXINT as both 
> baseOffset and offset are of type long. It was introduced in this [pr| 
> https://github.com/apache/kafka/pull/2210/files/56d1f8196b77a47b176b7bbd1e4220a3be827631]
> These were the outputs of dumping the first two log segments
> {noformat}
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0.log
> Dumping /kafka-logs/__consumer_offsets-12/.log
> Starting offset: 0
> offset: 1810054758 position: 0 NoTimestampType: -1 isvalid: true payloadsize: 
> -1 magic: 0 compresscodec: NONE crc: 3127861909 keysize: 34
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0002147343575.log
> Dumping /kafka-logs/__consumer_offsets-12/002147343575.log
> Starting offset: 2147343575
> offset: 2147539884 position: 0 NoTimestampType: -1 isvalid: true paylo
> adsize: -1 magic: 0 compresscodec: NONE crc: 2282192097 keysize: 34
> {noformat}
> My guess is that since 2147539884 is larger than MAXINT, we are hitting this 
> exception. Was there a specific reason, this check was added in 0.10.2?
> E.g. if the first offset is a key = "key 0" and then we have MAXINT + 1 of 
> "key 1" following, wouldn't we run into this situation whenever the log 
> cleaner runs?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3326: MINOR: Add unit tests for PluginDesc in Connect.

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3326


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Resolved] (KAFKA-5317) Update KIP-98 to reflect changes during implementation.

[Apurva Mehta (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5317?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Apurva Mehta resolved KAFKA-5317.
-
Resolution: Fixed

> Update KIP-98 to reflect changes during implementation.
> ---
>
> Key: KAFKA-5317
> URL: https://issues.apache.org/jira/browse/KAFKA-5317
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Apurva Mehta
>Assignee: Apurva Mehta
>Priority: Blocker
>  Labels: documentation, exactly-once
> Fix For: 0.11.0.0
>
>
> While implementing the EOS design, there are some minor (or major?) tweaks we 
> made as we hands-on the code bases. We will compile all these changes in a 
> single run at the end of the code-ready and this JIRA is for keeping track of 
> all the changes we made.
> 02/27/2017: collapse the two types of transactional log messages into a 
> single type with key as transactional id, and value as [pid, epoch, 
> transaction_timeout, transaction_state, [topic partition ] ]. Also using a 
> single memory map in cache instead of two on the TC.
> 03/01/2017: for pid expiration, we decided to use min(transactional id 
> expiration timeout, topic retention). For topics enabled for compaction only, 
> we just use the transactional timeout. If the retention setting is larger 
> than the transactional id expiration timeout, then the pid will be 
> "logically" expired (i.e. we will remove it from the cached pid mapping and 
> ignore it when rebuilding the cache from the log)
> 03/20/2017: add a new exception type in `o.a.k.common.errors` for invalid 
> transaction timeout values.
> 03/25/2017: extend WriteTxnMarkerRequest to contain multiple markers for 
> multiple PIDs with a single request.
> 04/20/2017: add transactionStartTime to TransactionMetadata
> 04/26/2017: added a new retriable error: Errors.CONCURRENT_TRANSACTIONS
> 04/01/2017: We also enforce acks=all on the client when idempotence is 
> enabled. Without this, we cannot again guarantee idemptoence.
> 04/10/2017: WE also don't pass the underlying exception to 
> `RetriableOffsetCommitFailedException` anymore: 
> https://issues.apache.org/jira/browse/KAFKA-5052



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5317) Update KIP-98 to reflect changes during implementation.

[Apurva Mehta (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5317?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048515#comment-16048515
 ] 

Apurva Mehta commented on KAFKA-5317:
-

I updated both the wiki and the google doc to reflect the changes on this list. 
Closing this out.

> Update KIP-98 to reflect changes during implementation.
> ---
>
> Key: KAFKA-5317
> URL: https://issues.apache.org/jira/browse/KAFKA-5317
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Apurva Mehta
>Assignee: Apurva Mehta
>Priority: Blocker
>  Labels: documentation, exactly-once
> Fix For: 0.11.0.0
>
>
> While implementing the EOS design, there are some minor (or major?) tweaks we 
> made as we hands-on the code bases. We will compile all these changes in a 
> single run at the end of the code-ready and this JIRA is for keeping track of 
> all the changes we made.
> 02/27/2017: collapse the two types of transactional log messages into a 
> single type with key as transactional id, and value as [pid, epoch, 
> transaction_timeout, transaction_state, [topic partition ] ]. Also using a 
> single memory map in cache instead of two on the TC.
> 03/01/2017: for pid expiration, we decided to use min(transactional id 
> expiration timeout, topic retention). For topics enabled for compaction only, 
> we just use the transactional timeout. If the retention setting is larger 
> than the transactional id expiration timeout, then the pid will be 
> "logically" expired (i.e. we will remove it from the cached pid mapping and 
> ignore it when rebuilding the cache from the log)
> 03/20/2017: add a new exception type in `o.a.k.common.errors` for invalid 
> transaction timeout values.
> 03/25/2017: extend WriteTxnMarkerRequest to contain multiple markers for 
> multiple PIDs with a single request.
> 04/20/2017: add transactionStartTime to TransactionMetadata
> 04/26/2017: added a new retriable error: Errors.CONCURRENT_TRANSACTIONS
> 04/01/2017: We also enforce acks=all on the client when idempotence is 
> enabled. Without this, we cannot again guarantee idemptoence.
> 04/10/2017: WE also don't pass the underlying exception to 
> `RetriableOffsetCommitFailedException` anymore: 
> https://issues.apache.org/jira/browse/KAFKA-5052



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5052) We shouldn't pass the underlying exception to RetriableCommitFailedException when an async offset commit fails.

[Apurva Mehta (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5052?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Apurva Mehta updated KAFKA-5052:

Description: 
* This is a follow up to : https://issues.apache.org/jira/browse/KAFKA-3527

We currently wrap retriable exceptions encountered during offset commits in a 
`RetriableOffsetCommitException`. The problem is that today we also pass the 
underlying internal exception on to the user. There isn't a really good reason 
to do this, since the user will not handle each individual exception 
differently: they will just retry anyway.

We should not pass on the underlying internal exception. It makes the API 
simpler, and also allows us to change things underneath with more flexibility.

  was:
This is a follow up to : https://issues.apache.org/jira/browse/KAFKA-3527

We currently wrap retriable exceptions encountered during offset commits in a 
`RetriableOffsetCommitException`. The problem is that today we also pass the 
underlying internal exception on to the user. There isn't a really good reason 
to do this, since the user will not handle each individual exception 
differently: they will just retry anyway.

We should not pass on the underlying internal exception. It makes the API 
simpler, and also allows us to change things underneath with more flexibility.


> We shouldn't pass the underlying exception to RetriableCommitFailedException 
> when an async offset commit fails.
> ---
>
> Key: KAFKA-5052
> URL: https://issues.apache.org/jira/browse/KAFKA-5052
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Apurva Mehta
>Assignee: Apurva Mehta
> Fix For: 0.11.0.0
>
>
> * This is a follow up to : https://issues.apache.org/jira/browse/KAFKA-3527
> We currently wrap retriable exceptions encountered during offset commits in a 
> `RetriableOffsetCommitException`. The problem is that today we also pass the 
> underlying internal exception on to the user. There isn't a really good 
> reason to do this, since the user will not handle each individual exception 
> differently: they will just retry anyway.
> We should not pass on the underlying internal exception. It makes the API 
> simpler, and also allows us to change things underneath with more flexibility.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5438) UnsupportedOperationException in WriteTxnMarkers handler

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5438?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048514#comment-16048514
 ] 

ASF GitHub Bot commented on KAFKA-5438:
---

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3313


> UnsupportedOperationException in WriteTxnMarkers handler
> 
>
> Key: KAFKA-5438
> URL: https://issues.apache.org/jira/browse/KAFKA-5438
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Jason Gustafson
>Assignee: Apurva Mehta
>Priority: Blocker
>  Labels: exactly-once
> Fix For: 0.11.0.0
>
>
> {code}
> [2017-06-10 19:16:36,280] ERROR [KafkaApi-2] Error when handling request 
> {replica_id=1,max_wait_time=500,min_bytes=1,max_bytes=10485760,isolation_level=0,topics=[{topic=__transaction_state,partitions=[{partition=45,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=15,f
> etch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=20,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=3,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=21,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=29,fetch_offset=0,lo
> g_start_offset=0,max_bytes=1048576},{partition=39,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=38,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=9,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=33,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=32,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=17,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=23,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=47,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=26,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=5,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=27,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=41,fetch_offset=48,log_start_offset=0,max_bytes=1048576},{partition=35,fetch_offset=0,log_start_offset=0,max_bytes=1048576}]},{topic=__consumer_offsets,partitions=[{partition=8,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=21,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=27,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=9,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=41,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=33,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=23,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=47,fetch_offset=41,log_start_offset=0,max_bytes=1048576},{partition=3,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=15,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=17,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=11,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=14,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=20,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=39,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=45,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=5,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=26,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=29,fetch_offset=0,log_start_offset=0,max_bytes=1048576}]},{topic=output-topic,partitions=[{partition=1,fetch_offset=4522,log_start_offset=0,max_bytes=1048576}]}]}
>  (kafka.server.KafkaApis)
> java.lang.UnsupportedOperationException
> at java.util.AbstractMap.put(AbstractMap.java:203)
> at java.util.AbstractMap.putAll(AbstractMap.java:273)
> at 
> kafka.server.KafkaApis.kafka$server$KafkaApis$$sendResponseCallback$13(KafkaApis.scala:1509)
> at 
> kafka.server.KafkaApis$$anonfun$handleWriteTxnMarkersRequest$2$$anonfun$apply$20.apply(KafkaApis.scala:1571)
> at 
> kafka.server.KafkaApis$$anonfun$handleWriteTxnMarkersRequest$2$$anonfun$apply$20.apply(KafkaApis.scala:1571)
> at kafka.server.DelayedProduce.onComplete(DelayedProduce.scala:131)
> at 
> kafka.server.DelayedOperation.forceComplete(DelayedOperation.scala:66)
> at kafka.server.DelayedProduce.tryComplete(DelayedProduce.scala:113)
> at 
> kafka.server.DelayedProduce.safeTryComplete(DelayedProduce.scala:76)
> at 
> kafka.server.DelayedOperationPurgatory$Watchers.tryCompleteWatched(DelayedOperation.scala:338)
> at 
> kafka.server.DelayedOperationPurgatory.checkAndComplete(DelayedOperation.scala:244)
> at 
> kafka.server.ReplicaManager.tryCompleteDelayedProduce(ReplicaManager.scala:250)
> at 
> kafka.cluster.Parti

[GitHub] kafka pull request #3313: KAFKA-5438: Fix UnsupportedOperationException in w...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3313


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Resolved] (KAFKA-5438) UnsupportedOperationException in WriteTxnMarkers handler

[Jason Gustafson (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5438?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Jason Gustafson resolved KAFKA-5438.

Resolution: Fixed

Issue resolved by pull request 3313
[https://github.com/apache/kafka/pull/3313]

> UnsupportedOperationException in WriteTxnMarkers handler
> 
>
> Key: KAFKA-5438
> URL: https://issues.apache.org/jira/browse/KAFKA-5438
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Jason Gustafson
>Assignee: Apurva Mehta
>Priority: Blocker
>  Labels: exactly-once
> Fix For: 0.11.0.0
>
>
> {code}
> [2017-06-10 19:16:36,280] ERROR [KafkaApi-2] Error when handling request 
> {replica_id=1,max_wait_time=500,min_bytes=1,max_bytes=10485760,isolation_level=0,topics=[{topic=__transaction_state,partitions=[{partition=45,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=15,f
> etch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=20,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=3,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=21,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=29,fetch_offset=0,lo
> g_start_offset=0,max_bytes=1048576},{partition=39,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=38,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=9,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=33,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=32,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=17,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=23,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=47,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=26,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=5,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=27,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=41,fetch_offset=48,log_start_offset=0,max_bytes=1048576},{partition=35,fetch_offset=0,log_start_offset=0,max_bytes=1048576}]},{topic=__consumer_offsets,partitions=[{partition=8,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=21,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=27,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=9,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=41,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=33,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=23,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=47,fetch_offset=41,log_start_offset=0,max_bytes=1048576},{partition=3,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=15,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=17,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=11,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=14,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=20,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=39,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=45,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=5,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=26,fetch_offset=0,log_start_offset=0,max_bytes=1048576},{partition=29,fetch_offset=0,log_start_offset=0,max_bytes=1048576}]},{topic=output-topic,partitions=[{partition=1,fetch_offset=4522,log_start_offset=0,max_bytes=1048576}]}]}
>  (kafka.server.KafkaApis)
> java.lang.UnsupportedOperationException
> at java.util.AbstractMap.put(AbstractMap.java:203)
> at java.util.AbstractMap.putAll(AbstractMap.java:273)
> at 
> kafka.server.KafkaApis.kafka$server$KafkaApis$$sendResponseCallback$13(KafkaApis.scala:1509)
> at 
> kafka.server.KafkaApis$$anonfun$handleWriteTxnMarkersRequest$2$$anonfun$apply$20.apply(KafkaApis.scala:1571)
> at 
> kafka.server.KafkaApis$$anonfun$handleWriteTxnMarkersRequest$2$$anonfun$apply$20.apply(KafkaApis.scala:1571)
> at kafka.server.DelayedProduce.onComplete(DelayedProduce.scala:131)
> at 
> kafka.server.DelayedOperation.forceComplete(DelayedOperation.scala:66)
> at kafka.server.DelayedProduce.tryComplete(DelayedProduce.scala:113)
> at 
> kafka.server.DelayedProduce.safeTryComplete(DelayedProduce.scala:76)
> at 
> kafka.server.DelayedOperationPurgatory$Watchers.tryCompleteWatched(DelayedOperation.scala:338)
> at 
> kafka.server.DelayedOperationPurgatory.checkAndComplete(DelayedOperation.scala:244)
> at 
> kafka.server.ReplicaManager.tryCompleteDelayedProduce(ReplicaManager.scala:250)
> at 
> kafka.cluster.Partition.tryCompleteDelayedRequests(Partition.scal

[jira] [Updated] (KAFKA-5442) Streams producer `client.id` are not unique for EOS

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-5442:
---
Status: Patch Available  (was: In Progress)

> Streams producer `client.id` are not unique for EOS
> ---
>
> Key: KAFKA-5442
> URL: https://issues.apache.org/jira/browse/KAFKA-5442
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Blocker
> Fix For: 0.11.0.0
>
>
> Wth producer per task model in EOS, producer `client.id` must encode 
> `task.id` to make IDs unique. Currently, only thread-id is encoded resulting 
> in naming conflict.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5442) Streams producer `client.id` are not unique for EOS

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048509#comment-16048509
 ] 

ASF GitHub Bot commented on KAFKA-5442:
---

GitHub user mjsax opened a pull request:

https://github.com/apache/kafka/pull/3329

KAFKA-5442: Streams producer client.id are not unique for EOS



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/mjsax/kafka kafka-5442-producer-id-conflict

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3329.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3329


commit f1162dbe5f41d40f644189af5fab3bb171ed9444
Author: Matthias J. Sax 
Date:   2017-06-13T23:05:11Z

KAFKA-5442: Streams producer client.id are not unique for EOS




> Streams producer `client.id` are not unique for EOS
> ---
>
> Key: KAFKA-5442
> URL: https://issues.apache.org/jira/browse/KAFKA-5442
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Blocker
> Fix For: 0.11.0.0
>
>
> Wth producer per task model in EOS, producer `client.id` must encode 
> `task.id` to make IDs unique. Currently, only thread-id is encoded resulting 
> in naming conflict.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3329: KAFKA-5442: Streams producer client.id are not uni...

[mjsax]

GitHub user mjsax opened a pull request:

https://github.com/apache/kafka/pull/3329

KAFKA-5442: Streams producer client.id are not unique for EOS



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/mjsax/kafka kafka-5442-producer-id-conflict

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3329.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3329


commit f1162dbe5f41d40f644189af5fab3bb171ed9444
Author: Matthias J. Sax 
Date:   2017-06-13T23:05:11Z

KAFKA-5442: Streams producer client.id are not unique for EOS




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] kafka pull request #3328: HOTFIX: log level should be ERROR for conflicting ...

[mjsax]

GitHub user mjsax opened a pull request:

https://github.com/apache/kafka/pull/3328

HOTFIX: log level should be ERROR for conflicting metric names



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/mjsax/kafka hotfix-app-info-log

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3328.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3328


commit 5b0724b90f44ad6f2e073138fcf34f5a5dfee392
Author: Matthias J. Sax 
Date:   2017-06-13T22:54:15Z

HOTFIX: log level should be ERROR for conflicting metric names




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] kafka pull request #3327: HOTFIX: Introduce max wait time for retry-and-back...

[mjsax]

GitHub user mjsax opened a pull request:

https://github.com/apache/kafka/pull/3327

HOTFIX: Introduce max wait time for retry-and-backoff while creating tasks



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/mjsax/kafka hotfix-backoff-retry

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3327.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3327


commit 4e7dfd19ed6917e7b5f45f725c3a01f7e82bfa82
Author: Matthias J. Sax 
Date:   2017-06-13T22:50:24Z

HOTFIX: Introduce max wait time for retry-and-backoff while creating tasks




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Work started] (KAFKA-5442) Streams producer `client.id` are not unique for EOS

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5442?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Work on KAFKA-5442 started by Matthias J. Sax.
--
> Streams producer `client.id` are not unique for EOS
> ---
>
> Key: KAFKA-5442
> URL: https://issues.apache.org/jira/browse/KAFKA-5442
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.11.0.0
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Blocker
> Fix For: 0.11.0.0
>
>
> Wth producer per task model in EOS, producer `client.id` must encode 
> `task.id` to make IDs unique. Currently, only thread-id is encoded resulting 
> in naming conflict.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (KAFKA-5442) Streams producer `client.id` are not unique for EOS

[Matthias J. Sax (JIRA)]

Matthias J. Sax created KAFKA-5442:
--

 Summary: Streams producer `client.id` are not unique for EOS
 Key: KAFKA-5442
 URL: https://issues.apache.org/jira/browse/KAFKA-5442
 Project: Kafka
  Issue Type: Bug
  Components: streams
Affects Versions: 0.11.0.0
Reporter: Matthias J. Sax
Assignee: Matthias J. Sax
Priority: Blocker
 Fix For: 0.11.0.0


Wth producer per task model in EOS, producer `client.id` must encode `task.id` 
to make IDs unique. Currently, only thread-id is encoded resulting in naming 
conflict.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Comment Edited] (KAFKA-5348) kafka-consumer-groups.sh refuses to remove groups without ids

[Vahid Hashemian (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048486#comment-16048486
 ] 

Vahid Hashemian edited comment on KAFKA-5348 at 6/13/17 10:36 PM:
--

[~bobrik] In the scenario you described I assume some consumer id exists under 
the {{/ids}} path. By design, the consumer group (for old consumers) can be 
deleted only if there is no active consumer in the group. There are active 
consumers group in the group iff the path {{/ids}} exists for this group and 
there are consumer ids inside this path. If this is not what you're 
experiencing please advise and perhaps provide steps to reproduce. Thanks.



was (Author: vahid):
[~bobrik] In the scenario you described I assume some consumer id exists under 
the {{/ids}} path. By design, the consumer group (for old consumers) can be 
deleted only if there is no active consumer in the group. There is no active 
consumer group in the group iff the path {{/ids}} exists for this group and 
there are consumer ids inside this path. If this is not what you're 
experiencing please advise and perhaps provide steps to reproduce. Thanks.


> kafka-consumer-groups.sh refuses to remove groups without ids
> -
>
> Key: KAFKA-5348
> URL: https://issues.apache.org/jira/browse/KAFKA-5348
> Project: Kafka
>  Issue Type: Bug
>  Components: admin
>Affects Versions: 0.10.2.0
>Reporter: Ivan Babrou
>Assignee: Vahid Hashemian
>
> In zookeeper I have:
> {noformat}
> [zk: foo(CONNECTED) 37] ls /kafka/logs/consumers/console-consumer-4107
> [offsets]
> {noformat}
> This consumer group also shows up when I list consumer groups:
> {noformat}
> $ /usr/local/kafka/bin/kafka-consumer-groups.sh --zookeeper 
> foo:2181/kafka/logs --list | fgrep console-consumer-4107
> Note: This will only show information about consumers that use ZooKeeper (not 
> those using the Java consumer API).
> console-consumer-4107
> {noformat}
> But I cannot remove this group:
> {noformat}
> $ /usr/local/kafka/bin/kafka-consumer-groups.sh --zookeeper 
> 36zk1.in.pdx.cfdata.org:2181/kafka/logs --delete --group console-consumer-4107
> Note: This will only show information about consumers that use ZooKeeper (not 
> those using the Java consumer API).
> Error: Delete for group 'console-consumer-4107' failed because group does not 
> exist.
> {noformat}
> I ran tcpdump and it turns out that /ids path is checked:
> {noformat}
> $.e.P.fP...&..<...//kafka/logs/consumers/console-consumer-4107/ids.
> {noformat}
> I think kafka should not check for /ids, it should check for / instead here.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3326: MINOR: Add unit tests for PluginDesc in Connect.

[kkonstantine]

GitHub user kkonstantine opened a pull request:

https://github.com/apache/kafka/pull/3326

MINOR: Add unit tests for PluginDesc in Connect.

Related to https://github.com/apache/kafka/pull/3321

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/kkonstantine/kafka 
MINOR-Add-tests-for-PluginDesc

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3326.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3326


commit ba6425846602d07f7271c5e2399a1abcf5236ed4
Author: Konstantine Karantasis 
Date:   2017-06-13T22:27:28Z

MINOR: Add unit tests for PluginDesc in Connect.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-5348) kafka-consumer-groups.sh refuses to remove groups without ids

[Vahid Hashemian (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048486#comment-16048486
 ] 

Vahid Hashemian commented on KAFKA-5348:


[~bobrik] In the scenario you described I assume some consumer id exists under 
the {{/ids}} path. By design, the consumer group (for old consumers) can be 
deleted only if there is no active consumer in the group. There is no active 
consumer group in the group iff the path {{/ids}} exists for this group and 
there are consumer ids inside this path. If this is not what you're 
experiencing please advise and perhaps provide steps to reproduce. Thanks.


> kafka-consumer-groups.sh refuses to remove groups without ids
> -
>
> Key: KAFKA-5348
> URL: https://issues.apache.org/jira/browse/KAFKA-5348
> Project: Kafka
>  Issue Type: Bug
>  Components: admin
>Affects Versions: 0.10.2.0
>Reporter: Ivan Babrou
>Assignee: Vahid Hashemian
>
> In zookeeper I have:
> {noformat}
> [zk: foo(CONNECTED) 37] ls /kafka/logs/consumers/console-consumer-4107
> [offsets]
> {noformat}
> This consumer group also shows up when I list consumer groups:
> {noformat}
> $ /usr/local/kafka/bin/kafka-consumer-groups.sh --zookeeper 
> foo:2181/kafka/logs --list | fgrep console-consumer-4107
> Note: This will only show information about consumers that use ZooKeeper (not 
> those using the Java consumer API).
> console-consumer-4107
> {noformat}
> But I cannot remove this group:
> {noformat}
> $ /usr/local/kafka/bin/kafka-consumer-groups.sh --zookeeper 
> 36zk1.in.pdx.cfdata.org:2181/kafka/logs --delete --group console-consumer-4107
> Note: This will only show information about consumers that use ZooKeeper (not 
> those using the Java consumer API).
> Error: Delete for group 'console-consumer-4107' failed because group does not 
> exist.
> {noformat}
> I ran tcpdump and it turns out that /ids path is checked:
> {noformat}
> $.e.P.fP...&..<...//kafka/logs/consumers/console-consumer-4107/ids.
> {noformat}
> I think kafka should not check for /ids, it should check for / instead here.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5441) Fix transaction marker grouping by producerId in TC

[Apurva Mehta (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5441?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Apurva Mehta updated KAFKA-5441:

Labels: exactly-once  (was: )

> Fix transaction marker grouping by producerId in TC
> ---
>
> Key: KAFKA-5441
> URL: https://issues.apache.org/jira/browse/KAFKA-5441
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Jason Gustafson
>  Labels: exactly-once
> Fix For: 0.11.0.1
>
>
> It seems in some cases the WriteTxnMarker request can be sent with multiple 
> entries for the same ProducerId. This is the cause of KAFKA-5438. This is not 
> necessarily a correctness problem, but it seems unintentional and should 
> probably be fixed.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (KAFKA-5441) Fix transaction marker grouping by producerId in TC

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-5441:
--

 Summary: Fix transaction marker grouping by producerId in TC
 Key: KAFKA-5441
 URL: https://issues.apache.org/jira/browse/KAFKA-5441
 Project: Kafka
  Issue Type: Sub-task
Reporter: Jason Gustafson
 Fix For: 0.11.0.1


It seems in some cases the WriteTxnMarker request can be sent with multiple 
entries for the same ProducerId. This is the cause of KAFKA-5438. This is not 
necessarily a correctness problem, but it seems unintentional and should 
probably be fixed.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-4585) Offset fetch and commit requests use the same permissions

[Vahid Hashemian (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4585?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Vahid Hashemian updated KAFKA-4585:
---
Labels: kip  (was: needs-kip)

> Offset fetch and commit requests use the same permissions
> -
>
> Key: KAFKA-4585
> URL: https://issues.apache.org/jira/browse/KAFKA-4585
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.10.1.1
>Reporter: Ewen Cheslack-Postava
>Assignee: Vahid Hashemian
>  Labels: kip
>
> Currently the handling of permissions for consumer groups seems a bit odd 
> because most of the requests use the Read permission on the Group (join, 
> sync, heartbeat, leave, offset commit, and offset fetch). This means you 
> cannot lock down certain functionality for certain users. For this issue I'll 
> highlight a realistic issue since conflating the ability to perform most of 
> these operations may not be a serious issue.
> In particular, if you want tooling for monitoring offsets (i.e. you want to 
> be able to read from all groups) but don't want that tool to be able to write 
> offsets, you currently cannot achieve this. Part of the reason this seems odd 
> to me is that any operation which can mutate state seems like it should be a 
> Write operation (i.e. joining, syncing, leaving, and committing; maybe 
> heartbeat as well). However, [~hachikuji] has mentioned that the use of Read 
> may have been intentional. If that is the case, changing at least offset 
> fetch to be a Describe operation instead would allow isolating the mutating 
> vs non-mutating request types.
> Note that this would require a KIP and would potentially have some 
> compatibility implications. Note however, that if we went with the Describe 
> option, Describe is allowed by default when Read, Write, or Delete are 
> allowed, so this may not have to have any compatibility issues (if the user 
> previously allowed Read, they'd still have all the same capabilities as 
> before).



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Work started] (KAFKA-5370) Replace uses of old consumer with the new consumer

[Vahid Hashemian (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5370?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Work on KAFKA-5370 started by Vahid Hashemian.
--
> Replace uses of old consumer with the new consumer 
> ---
>
> Key: KAFKA-5370
> URL: https://issues.apache.org/jira/browse/KAFKA-5370
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Vahid Hashemian
>Assignee: Vahid Hashemian
>Priority: Minor
>
> Where possible, use the new consumer In tools and tests instead of the old 
> consumer, and remove the deprecation warning.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Assigned] (KAFKA-5212) Consumer ListOffsets request can starve group heartbeats

[Jason Gustafson (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5212?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Jason Gustafson reassigned KAFKA-5212:
--

Assignee: (was: Jason Gustafson)

> Consumer ListOffsets request can starve group heartbeats
> 
>
> Key: KAFKA-5212
> URL: https://issues.apache.org/jira/browse/KAFKA-5212
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Reporter: Jason Gustafson
> Fix For: 0.11.1.0
>
>
> The consumer is not able to send heartbeats while it is awaiting a 
> ListOffsets response. Typically this is not a problem because ListOffsets 
> requests are handled quickly, but in the worst case if the request takes 
> longer than the session timeout, the consumer will fall out of the group.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3321: HOTFIX: Handle Connector version returning 'null' ...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3321


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [DISCUSS] KIP-163: Lower the Minimum Required ACL Permission of OffsetFetch

[Vahid S Hashemian]

Hi Michal,

Thanks a lot for your feedback.

Your statement about Heartbeat is fair and makes sense. I'll update the 
KIP accordingly.

--Vahid




From:   Michal Borowiecki 
To: us...@kafka.apache.org, Vahid S Hashemian 
, dev@kafka.apache.org
Date:   06/13/2017 01:35 AM
Subject:Re: [DISCUSS] KIP-163: Lower the Minimum Required ACL 
Permission of OffsetFetch



Hi Vahid,
+1 wrt OffsetFetch.
The "Additional Food for Thought" mentions Heartbeat as a non-mutating 
action. I don't think that's true as the GroupCoordinator updates the 
latestHeartbeat field for the member and adds a new object to the 
heartbeatPurgatory, see completeAndScheduleNextHeartbeatExpiration() 
called from handleHeartbeat()

NB added dev mailing list back into CC as it seems to have been lost along 
the way.
Cheers,
Michał

On 12/06/17 18:47, Vahid S Hashemian wrote:
Hi Colin,

Thanks for the feedback.

To be honest, I'm not sure either why Read was selected instead of Write 
for mutating APIs in the initial design (I asked Ewen on the corresponding 

JIRA and he seemed unsure too).
Perhaps someone who was involved in the design can clarify.

Thanks.
--Vahid




From:   Colin McCabe 
To: us...@kafka.apache.org
Date:   06/12/2017 10:11 AM
Subject:Re: [DISCUSS] KIP-163: Lower the Minimum Required ACL 
Permission of OffsetFetch



Hi Vahid,

I think you make a valid point that the ACLs controlling group
operations are not very intuitive.

This is probably a dumb question, but why are we using Read for mutating
APIs?  Shouldn't that be Write?

The distinction between Describe and Read makes a lot of sense for
Topics.  A group isn't really something that you "read" from in the same
way as a topic, so it always felt kind of weird there.

best,
Colin


On Thu, Jun 8, 2017, at 11:29, Vahid S Hashemian wrote:

Hi all,

I'm resending my earlier note hoping it would spark some conversation
this 
time around :)

Thanks.
--Vahid




From:   "Vahid S Hashemian" 
To: dev , "Kafka User" 



Date:   05/30/2017 08:33 AM
Subject:KIP-163: Lower the Minimum Required ACL Permission of 
OffsetFetch



Hi,

I started a new KIP to improve the minimum required ACL permissions of 
some of the APIs: 


https://cwiki.apache.org/confluence/display/KAFKA/KIP-163%3A+Lower+the+Minimum+Required+ACL+Permission+of+OffsetFetch




The KIP is to address KAFKA-4585.

Feedback and suggestions are welcome!

Thanks.
--Vahid














-- 

Michal Borowiecki
Senior Software Engineer L4


T: 
+44 208 742 1600


+44 203 249 8448


 

E: 
michal.borowie...@openbet.com

W: 
www.openbet.com



OpenBet Ltd

Chiswick Park Building 9

566 Chiswick High Rd

London

W4 5XT

UK




This message is confidential and intended only for the addressee. If you 
have received this message in error, please immediately notify the 
postmas...@openbet.com and delete it from your system as well as any 
copies. The content of e-mails as well as traffic data may be monitored by 
OpenBet for employment and security purposes. To protect the environment 
please do not print this e-mail unless necessary. OpenBet Ltd. Registered 
Office: Chiswick Park Building 9, 566 Chiswick High Road, London, W4 5XT, 
United Kingdom. A company registered in England and Wales. Registered no. 
3134634. VAT no. GB927523612

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

[Bill Bejeck]

+1 on deferring discussion on Serdes until API improvements are ironed out.

On Tue, Jun 13, 2017 at 2:06 PM, Matthias J. Sax 
wrote:

> Hi,
>
> I am just catching up on this thread. (1) as most people agree, we
> should not add anything to KStreamBuilder (btw: we actually plan to move
> #merge() to KStream and deprecate it on KStreamBuilder as it's a quite
> unnatural API atm).
>
> About specifying Serdes: there is still the idea to improve to overall
> API from the current "we are adding more overloads"-pattern to a
> builder-like pattern. This might make the whole discussion void if we do
> this. Thus, it might make sense to keep this in mind (or even delay this
> KIP?). It seems a waste of time to discuss all this if we are going to
> chance it in 2 month anyway... Just saying.
>
>
> -Matthias
>
> On 6/13/17 8:05 AM, Michal Borowiecki wrote:
> > You're right, I haven't thought of that.
> >
> > Cheers,
> >
> > Michał
> >
> >
> > On 13/06/17 13:00, Kyle Winkelman wrote:
> >> First, I would prefer not calling it aggregate because there are already
> >> plenty of aggregate methods.
> >>
> >> Second, I dont think this would really work because after each aggregate
> >> you now have a unique KTable (someone may want a table with 4 streams
> and
> >> reuse those 4 in another table but with one more stream added) and
> unless
> >> we completely duplicate everything every time this isnt really possible.
> >> Additionally, the cogroup way just requires 1 more call to create two
> >> different tables (normal, windowed, and session windowed) this new way
> >> would require copying the aggregate chain.
> >>
> >> Another way to think about it is with cogroup we know that when they
> call
> >> aggregate they arent going to be adding any more aggregators to that
> table
> >> but your way requires us to assume they are done adding aggregators
> after
> >> each call so we must return a ktable just to possibly not need to have
> >> created it.
> >>
> >> On Jun 13, 2017 5:20 AM, "Michal Borowiecki" <
> michal.borowie...@openbet.com>
> >> wrote:
> >>
> >>> Actually, just had a thought. It started with naming.
> >>>
> >>> Are we actually co-grouping these streams or are we co-aggregating
> them?
> >>>
> >>> After all, in each of the cogroup calls we are providing an Aggregator
> >>> implementation.
> >>>
> >>>
> >>> If they are really co-aggregated, why don't we turn this around:
> >>> KGroupedStream grouped1 = builder.stream("topic1").
> groupByKey();
> >>> KGroupedStream grouped2 = builder.stream("topic2").
> groupByKey();
> >>> KGroupedStream grouped3 = builder.stream("topic3").
> groupByKey();
> >>>
> >>> KTable coagg = grouped1.aggregate(initializer1, aggregator1,
> >>> aggValueSerde1) // this is the unchanged aggregate method
> >>> .aggregate(grouped2, aggregator2)  // this is a new method
> >>> .aggregate(grouped3, aggregator3); // ditto
> >>>
> >>> This means instead of adding cogroup methods on KGroupStream interface,
> >>> adding aggregate method on KTable interface.
> >>>
> >>> Is that feasible?
> >>>
> >>> Cheers,
> >>> Michał
> >>>
> >>> On 13/06/17 10:56, Michal Borowiecki wrote:
> >>>
> >>> Also, I still feel that putting initializer on the first cogroup can
> >>> mislead users into thinking the first stream is in some way special.
> >>> Just my 5c.
> >>> Michał
> >>>
> >>> On 13/06/17 09:54, Michal Borowiecki wrote:
> >>>
> >>> Agree completely with the argument for serdes belonging in the same
> place
> >>> as the state store name, which is in the aggregate method.
> >>>
> >>> Cheers,
> >>>
> >>> Michał
> >>>
> >>> On 12/06/17 18:20, Xavier Léauté wrote:
> >>>
> >>> I think we are discussing two separate things here, so it might be
> worth
> >>> clarifying:
> >>>
> >>> 1) the position of the initializer with respect to the aggregators. If
> I
> >>> understand correctly, Guozhang seems to think it is more natural to
> specify
> >>> the initializer first, despite it not bearing any relation to the first
> >>> aggregator. I can see the argument for specifying the initializer
> first,
> >>> but I think it is debatable whether mixing it into the first cogroup
> call
> >>> leads to a cleaner API or not.
> >>>
> >>> 2) where the serde should be defined (if necessary). Looking at our
> >>> existing APIs in KGroupedStreams, we always offer two aggregate()
> >>> methods. The first one takes the name of the store and associated
> aggregate
> >>> value serde e.g. KGroupedStream.aggregate(Initializer initializer,
> >>> Aggregator aggregator, Serde
> aggValueSerde,
> >>> String queryableStoreName)
> >>> The second one only takes a state store supplier, and does not specify
> any
> >>> serde, e.g. KGroupedStream.aggregate(Initializer
> >>> initializer, Aggregator aggregator, final
> >>> StateStoreSupplier storeSupplier)
> >>> Presumably, when specifying a state store supplier it shouldn't be
> >>> necessary to specify an aggregate value serde, since the provided
> >>> statestore might not need to serialize

[jira] [Commented] (KAFKA-5413) Log cleaner fails due to large offset in segment file

[Jun Rao (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5413?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048429#comment-16048429
 ] 

Jun Rao commented on KAFKA-5413:


[~Kelvinrutt], yes, that seems to be the issue. segs.head.index.lastOffset 
doesn't give the precise last offset in the segment, especially in the case 
where the index is empty. The issue seems to be there even in 0.9.0.

About the fix. LogSegment.nextOffset() gives the offset that we want, but can 
be expensive. Perhaps we could use the next segment's base offset instead. 
[~Kelvinrutt], do you think you could patch it?

Until this is fixed, not sure if there is an easy way to get around the issue.

> Log cleaner fails due to large offset in segment file
> -
>
> Key: KAFKA-5413
> URL: https://issues.apache.org/jira/browse/KAFKA-5413
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.10.2.0, 0.10.2.1
> Environment: Ubuntu 14.04 LTS, Oracle Java 8u92, kafka_2.11-0.10.2.0
>Reporter: Nicholas Ngorok
>  Labels: reliability
> Attachments: .index.cleaned, 
> .log, .log.cleaned, 
> .timeindex.cleaned, 002147422683.log
>
>
> The log cleaner thread in our brokers is failing with the trace below
> {noformat}
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 0 in log __consumer_offsets-12 (largest timestamp Thu Jun 08 
> 15:48:59 PDT 2017) into 0, retaining deletes. (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 2147343575 in log __consumer_offsets-12 (largest timestamp 
> Thu Jun 08 15:49:06 PDT 2017) into 0, retaining deletes. 
> (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,834] ERROR {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Error due to  (kafka.log.LogCleaner)
> java.lang.IllegalArgumentException: requirement failed: largest offset in 
> message set can not be safely converted to relative offset.
> at scala.Predef$.require(Predef.scala:224)
> at kafka.log.LogSegment.append(LogSegment.scala:109)
> at kafka.log.Cleaner.cleanInto(LogCleaner.scala:478)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:405)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:401)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:401)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:363)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:362)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.clean(LogCleaner.scala:362)
> at 
> kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:241)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:220)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
> [2017-06-08 15:49:54,835] INFO {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Stopped  (kafka.log.LogCleaner)
> {noformat}
> This seems to point at the specific line [here| 
> https://github.com/apache/kafka/blob/0.11.0/core/src/main/scala/kafka/log/LogSegment.scala#L92]
>  in the kafka src where the difference is actually larger than MAXINT as both 
> baseOffset and offset are of type long. It was introduced in this [pr| 
> https://github.com/apache/kafka/pull/2210/files/56d1f8196b77a47b176b7bbd1e4220a3be827631]
> These were the outputs of dumping the first two log segments
> {noformat}
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0.log
> Dumping /kafka-logs/__consumer_offsets-12/.log
> Starting offset: 0
> offset: 1810054758 position: 0 NoTimestampType: -1 isvalid: true payloadsize: 
> -1 magic: 0 compresscodec: NONE crc: 3127861909 keysize: 34
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0002147343575.log
> Dumping /kafka-logs/__consumer_offsets-12/002147343575.log
> Starting offset: 2147343575
> offset: 2147539884 position: 0 NoTimestampType: -1 isvalid: true paylo
> adsize: -1 magic: 0 compresscodec: NONE crc: 2282192097 keysize: 34
> {noformat}
> My guess is that since 2147539884 is larger than MAXINT, we are hitting this 
> exception. Was there a specific reason, this check was added in 0.10.2?
> E.g. if the first offset is a key = "key 0" and then we have MAXINT + 1 of 
> "key 1" following, wouldn't we run into this situation whenever the log 
> cleaner runs?



--

[DISCUSS] KIP-167: Add interface for the state store restoration process

[Bill Bejeck]

The KIP has been updated to incorporate support for users to receive
notification of state store restore progress for either pre-existing or
custom state stores.

A PR has been submitted concurrently
https://github.com/apache/kafka/pull/3325

Thanks,
Bill

Re: [DISCUSS]: KIP-159: Introducing Rich functions to Streams

[Jeyhun Karimov]

Thanks for the comment Matthias. After all the discussion (thanks to all
participants), I think this (single method that passes in a RecordContext
object) is the best alternative.
Just a side note: I think KAFKA-3907 [1] can also be integrated into the
KIP by adding related method inside RecordContext interface.


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


Cheers,
Jeyhun

On Tue, Jun 13, 2017 at 7:50 PM Matthias J. Sax 
wrote:

> Hi,
>
> I would like to push this discussion further. It seems we got nice
> alternatives (thanks for the summary Jeyhun!).
>
> With respect to RichFunctions and allowing them to be stateful, I have
> my doubt as expressed already. From my understanding, the idea was to
> give access to record metadata information only. If you want to do a
> stateful computation you should rather use #transform().
>
> Furthermore, as pointed out, we would need to switch to a
> supplier-pattern introducing many more overloads.
>
> For those reason, I advocate for a simple interface with a single method
> that passes in a RecordContext object.
>
>
> -Matthias
>
>
> On 6/6/17 5:15 PM, Guozhang Wang wrote:
> > Thanks for the comprehensive summary!
> >
> > Personally I'd prefer the option of passing RecordContext as an
> additional
> > parameter into he overloaded function. But I'm also open to other
> arguments
> > if there are sth. that I have overlooked.
> >
> > Guozhang
> >
> >
> > On Mon, Jun 5, 2017 at 3:19 PM, Jeyhun Karimov 
> wrote:
> >
> >> Hi,
> >>
> >> Thanks for your comments Matthias and Guozhang.
> >>
> >> Below I mention the quick summary of the main alternatives we looked at
> to
> >> introduce the Rich functions (I will refer to it as Rich functions
> until we
> >> find better/another name). Initially the proposed alternatives was not
> >> backwards-compatible, so I will not mention them.
> >> The related discussions are spread in KIP-149 and in this KIP (KIP-159)
> >> discussion threads.
> >>
> >>
> >>
> >> 1. The idea of rich functions came into the stage with KIP-149, in
> >> discussion thread. As a result we extended KIP-149 to support Rich
> >> functions as well.
> >>
> >> 2.  To as part of the Rich functions, we provided init
> (ProcessorContext)
> >> method. Afterwards, Dammian suggested that we should not provide
> >> ProcessorContext to users. As a result, we separated the two problems
> into
> >> two separate KIPs, as it seems they can be solved in parallel.
> >>
> >> - One approach we considered was :
> >>
> >> public interface ValueMapperWithKey {
> >> VR apply(final K key, final V value);
> >> }
> >>
> >> public interface RichValueMapper extends RichFunction{
> >> }
> >>
> >> public interface RichFunction {
> >> void init(RecordContext recordContext);
> >> void close();
> >> }
> >>
> >> public interface RecordContext {
> >> String applicationId();
> >> TaskId taskId();
> >> StreamsMetrics metrics();
> >> String topic();
> >> int partition();
> >> long offset();
> >> long timestamp();
> >> Map appConfigs();
> >> Map appConfigsWithPrefix(String prefix);
> >> }
> >>
> >>
> >> public interface ProcessorContext extends RecordContext {
> >>// all methods but the ones in RecordContext
> >> }
> >>
> >> As a result:
> >> * . All "withKey" and "withoutKey" interfaces can be converted to their
> >> Rich counterparts (with empty init() and close() methods)
> >> *. All related Processors will accept Rich interfaces in their
> >> constructors.
> >> *. So, we convert the related "withKey" or "withoutKey" interfaces to
> Rich
> >> interface while building the topology and initialize the related
> processors
> >> with Rich interfaces only.
> >> *. We will not need to overloaded methods for rich functions as Rich
> >> interfaces extend withKey interfaces. We will just check the object type
> >> and act accordingly.
> >>
> >>
> >>
> >>
> >> 3. There was some thoughts that the above approach does not support
> lambdas
> >> so we should support only one method, only init(RecordContext), as part
> of
> >> Rich interfaces.
> >> This is still in discussion. Personally I think Rich interfaces are by
> >> definition lambda-free and we should not care much about it.
> >>
> >>
> >> 4. Thanks to Matthias's discussion, an alternative we considered was to
> >> pass in the RecordContext as method parameter.  This might even allow to
> >> use Lambdas and we could keep the name RichFunction as we preserve the
> >> nature of being a function.
> >> "If you go with `init()` and `close()` we basically
> >> allow users to have an in-memory state for a function. Thus, we cannot
> >> share a single instance of RichValueMapper (etc) over multiple tasks and
> >> we would need a supplier pattern similar to #transform(). And this would
> >> "break the flow" of the API, as (Rich)ValueMapperSupplier would not
> >> inherit from ValueMapper and thus we would need many new overload for
> >> KStream/KTable classes". (Copy paste from Matthias's email)
> >>
> >

[jira] [Commented] (KAFKA-5413) Log cleaner fails due to large offset in segment file

[Nicholas Ngorok (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5413?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048394#comment-16048394
 ] 

Nicholas Ngorok commented on KAFKA-5413:


This happened again today on another cluster we had upgraded ~3 weeks ago. 
[~junrao] could you please suggest a workaround if it exists to tackle this 
issue? It is a high priority for us since log cleaner threads and in turn 
compaction have stopped.

> Log cleaner fails due to large offset in segment file
> -
>
> Key: KAFKA-5413
> URL: https://issues.apache.org/jira/browse/KAFKA-5413
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.10.2.0, 0.10.2.1
> Environment: Ubuntu 14.04 LTS, Oracle Java 8u92, kafka_2.11-0.10.2.0
>Reporter: Nicholas Ngorok
>  Labels: reliability
> Attachments: .index.cleaned, 
> .log, .log.cleaned, 
> .timeindex.cleaned, 002147422683.log
>
>
> The log cleaner thread in our brokers is failing with the trace below
> {noformat}
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 0 in log __consumer_offsets-12 (largest timestamp Thu Jun 08 
> 15:48:59 PDT 2017) into 0, retaining deletes. (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,822] INFO {kafka-log-cleaner-thread-0} Cleaner 0: 
> Cleaning segment 2147343575 in log __consumer_offsets-12 (largest timestamp 
> Thu Jun 08 15:49:06 PDT 2017) into 0, retaining deletes. 
> (kafka.log.LogCleaner)
> [2017-06-08 15:49:54,834] ERROR {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Error due to  (kafka.log.LogCleaner)
> java.lang.IllegalArgumentException: requirement failed: largest offset in 
> message set can not be safely converted to relative offset.
> at scala.Predef$.require(Predef.scala:224)
> at kafka.log.LogSegment.append(LogSegment.scala:109)
> at kafka.log.Cleaner.cleanInto(LogCleaner.scala:478)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:405)
> at 
> kafka.log.Cleaner$$anonfun$cleanSegments$1.apply(LogCleaner.scala:401)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.cleanSegments(LogCleaner.scala:401)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:363)
> at kafka.log.Cleaner$$anonfun$clean$4.apply(LogCleaner.scala:362)
> at scala.collection.immutable.List.foreach(List.scala:381)
> at kafka.log.Cleaner.clean(LogCleaner.scala:362)
> at 
> kafka.log.LogCleaner$CleanerThread.cleanOrSleep(LogCleaner.scala:241)
> at kafka.log.LogCleaner$CleanerThread.doWork(LogCleaner.scala:220)
> at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
> [2017-06-08 15:49:54,835] INFO {kafka-log-cleaner-thread-0} 
> [kafka-log-cleaner-thread-0], Stopped  (kafka.log.LogCleaner)
> {noformat}
> This seems to point at the specific line [here| 
> https://github.com/apache/kafka/blob/0.11.0/core/src/main/scala/kafka/log/LogSegment.scala#L92]
>  in the kafka src where the difference is actually larger than MAXINT as both 
> baseOffset and offset are of type long. It was introduced in this [pr| 
> https://github.com/apache/kafka/pull/2210/files/56d1f8196b77a47b176b7bbd1e4220a3be827631]
> These were the outputs of dumping the first two log segments
> {noformat}
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0.log
> Dumping /kafka-logs/__consumer_offsets-12/.log
> Starting offset: 0
> offset: 1810054758 position: 0 NoTimestampType: -1 isvalid: true payloadsize: 
> -1 magic: 0 compresscodec: NONE crc: 3127861909 keysize: 34
> :~$ /usr/bin/kafka-run-class kafka.tools.DumpLogSegments --deep-iteration 
> --files /kafka-logs/__consumer_offsets-12/000
> 0002147343575.log
> Dumping /kafka-logs/__consumer_offsets-12/002147343575.log
> Starting offset: 2147343575
> offset: 2147539884 position: 0 NoTimestampType: -1 isvalid: true paylo
> adsize: -1 magic: 0 compresscodec: NONE crc: 2282192097 keysize: 34
> {noformat}
> My guess is that since 2147539884 is larger than MAXINT, we are hitting this 
> exception. Was there a specific reason, this check was added in 0.10.2?
> E.g. if the first offset is a key = "key 0" and then we have MAXINT + 1 of 
> "key 1" following, wouldn't we run into this situation whenever the log 
> cleaner runs?



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5363) Add ability to batch restore and receive restoration stats.

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5363?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048378#comment-16048378
 ] 

ASF GitHub Bot commented on KAFKA-5363:
---

GitHub user bbejeck opened a pull request:

https://github.com/apache/kafka/pull/3325

KAFKA-5363 [WIP] : Initial cut at implementing bulk load for persistent 
stat…

…e stores

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/bbejeck/kafka 
KAFKA-5363_add_ability_to_batch_restore

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3325.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3325


commit 012cc64568ed95fc8cc240536f645179aa3a7cec
Author: Bill Bejeck 
Date:   2017-06-06T21:49:25Z

KAFKA-5363: Initial cut at implementing bulk load for persistent state 
stores

KAFKA-5363: Initial cut at implementing bulk load for persistent state 
stores

KAFKA-5363: added state store recovery benchmark

KAFKA-5363: remove unused import.

KAFKA-5363: Separated out listening to restoration events in separate 
interface, allows for notifications from current state stores and custom state 
stores.

KAFKA-5363: minor cleanup




> Add ability to batch restore and receive restoration stats.
> ---
>
> Key: KAFKA-5363
> URL: https://issues.apache.org/jira/browse/KAFKA-5363
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Bill Bejeck
>Assignee: Bill Bejeck
>  Labels: kip
> Fix For: 0.11.1.0
>
>
> Currently, when restoring a state store in a Kafka Streams application, we 
> put one key-value at a time into the store.  
> This task aims to make this recovery more efficient by creating a new 
> interface with "restoreAll" functionality allowing for bulk writes by the 
> underlying state store implementation.  
> The proposal will also add "beginRestore" and "endRestore" callback methods 
> potentially used for 
> Tracking when the bulk restoration process begins and ends.
> Keeping track of the number of records and last offset restored.
> KIP: 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-167%3A+Add+interface+for+the+state+store+restoration+process



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3325: KAFKA-5363 [WIP] : Initial cut at implementing bul...

[bbejeck]

GitHub user bbejeck opened a pull request:

https://github.com/apache/kafka/pull/3325

KAFKA-5363 [WIP] : Initial cut at implementing bulk load for persistent 
stat…

…e stores

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/bbejeck/kafka 
KAFKA-5363_add_ability_to_batch_restore

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3325.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3325


commit 012cc64568ed95fc8cc240536f645179aa3a7cec
Author: Bill Bejeck 
Date:   2017-06-06T21:49:25Z

KAFKA-5363: Initial cut at implementing bulk load for persistent state 
stores

KAFKA-5363: Initial cut at implementing bulk load for persistent state 
stores

KAFKA-5363: added state store recovery benchmark

KAFKA-5363: remove unused import.

KAFKA-5363: Separated out listening to restoration events in separate 
interface, allows for notifications from current state stores and custom state 
stores.

KAFKA-5363: minor cleanup




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] kafka pull request #3324: Temporarily remove the apiVersions API for 0.11

[cmccabe]

GitHub user cmccabe opened a pull request:

https://github.com/apache/kafka/pull/3324

Temporarily remove the apiVersions API for 0.11



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/cmccabe/kafka removeApiVersions

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3324.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3324


commit fd983ce96aadd3e4f4a9602bfdc6e5020d33aeb8
Author: Colin P. Mccabe 
Date:   2017-06-13T20:34:04Z

Temporarily remove the apiVersions API for 0.11




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [VOTE] KIP-117: Add a public AdminClient API for Kafka admin operations

[Colin McCabe]

Hi all,

We're getting close to the final 0.11 release and we still haven't
gotten the version retrieval API for AdminClient in, as described by
KAFKA-5214.  Let's put this off until the next release so that we have
some time to let it soak before shipping.  This will just mean that we
will remove AdminClient#apiVersions for now, until the next release.  It
will still be possible to get version information with the existing
kafka-broker-api-versions.sh script.

best,
Colin


On Sat, Jun 3, 2017, at 01:08, Ismael Juma wrote:
> Hi Colin,
> 
> Thanks for the feedback. Regarding the behaviour for brokers older than
> 0.11.0, I had gone for the Javadoc note because it made it possible to
> avoid the inefficiency of getting all topics for users who have disabled
> auto topic creation.
> 
> After some thought and discussion, I agree that keeping the behaviour
> consistent across broker versions is the better option, so the PR was
> updated to do that.
> 
> Ismael
> 
> On Mon, May 22, 2017 at 7:42 PM, Colin McCabe  wrote:
> 
> > > As you noted, though, we don't have a way to do this for the 0.10.x
> > > releases.  It seems a bit harsh to have such different behavior there.
> > > Is there a way that we can fudge this a bit so that it mostly works?
> > > For example, when communicating with 0.10.x brokers, describeTopics
> > > could do a MetadataRequest(topics=*) to filter out non-existent topics.
> > >
> > > This would obviously have a bit of a time-of-check, time-of-use race
> > > condition since we're making two calls.  And also a scalability problem
> > > since we're using topics=*.  Is it worth it to make the behavior saner
> > > on older brokers?  Or should we add a JavaDoc note and move on?
> > >
> > > best,
> > > Colin
> > >
> > >
> > > On Fri, May 19, 2017, at 05:46, Ismael Juma wrote:
> > > > Hi all,
> > > >
> > > > Feedback from people who tried the AdminClient is that auto topic
> > > > creation
> > > > during describe is unexpected and confusing. This is consistent with
> > the
> > > > reaction of most people when they learn that MetadataRequest can cause
> > > > topics to be created. We had assumed that we'd tackle this issue for
> > all
> > > > the clients as part of deprecation of server-side auto topic creation
> > in
> > > > favour of client-side auto-topic creation.
> > > >
> > > > However, it would be better to do the right thing for the AdminClient
> > > > from
> > > > the start. Users will be less confused and we won't have to deal with
> > > > compatibility concerns. Jason suggested a simple solution: make it
> > > > possible
> > > > to disallow auto topic creation when sending the metadata request. The
> > > > AdminClient would take advantage of this now (i.e. 0.11.0.0) while the
> > > > producer and consumer would retain the existing behaviour. In a
> > > > subsequent
> > > > release, we'll work out the details of how to move away from
> > server-side
> > > > auto topic creation for the producer and consumer (taking into account
> > > > the
> > > > compatibility impact).
> > > >
> > > > Because of the protocol change, this solution would only help in cases
> > > > where the AdminClient is describing topics from a 0.11.0.0 or newer
> > > > broker.
> > > >
> > > > I submitted a PR for this and it's small and straightforward:
> > > >
> > > > https://github.com/apache/kafka/pull/3098
> > > >
> > > > Thoughts?
> > > >
> > > > Ismael
> > > >
> > > > On Sat, Mar 25, 2017 at 1:25 AM, Colin McCabe 
> > wrote:
> > > >
> > > > > With binding +1 votes from Gwen Shapira, Sriram Subramanian, and
> > Grant
> > > > > Henke, and a non-binding vote from Dong Lin, the vote passes.  There
> > > > > were no +0 or -1 votes.  As mentioned earlier, the interface will be
> > > > > unstable at first and we will continue to evolve it.
> > > > >
> > > > > thanks,
> > > > > Colin McCabe
> > > > >
> > > > >
> > > > > On Wed, Mar 22, 2017, at 10:21, Colin McCabe wrote:
> > > > > > On Fri, Mar 17, 2017, at 10:50, Jun Rao wrote:
> > > > > > > Hi, Colin,
> > > > > > >
> > > > > > > Thanks for the KIP. Looks good overall. A few comments below.
> > > > > > >
> > > > > > > 1. Sometimes we return
> > > > > > > CompletableFuture>
> > > > > > > and some other times we return
> > > > > > > Map>
> > > > > > > , which doesn't seem consistent. Is that intentional?
> > > > > >
> > > > > > Yes, this is intentional.  We got feedback from some people that
> > they
> > > > > > wanted a single future that would fail if anything failed.  Other
> > people
> > > > > > wanted to be able to detect failures on individual elements of a
> > batch.
> > > > > > This API lets us have both (you just choose which future you want
> > to
> > > > > > wait on).
> > > > > >
> > > > > > >
> > > > > > > 2. We support batching in CreateTopic/DeleteTopic/ListTopic,
> > but not
> > > > > in
> > > > > > > DescribeTopic. Should we add batching in DescribeTopic to make it
> > > > > > > consistent?
> > > > > >
> > > > > > Good idea.  Let's add batching to Des

Re: [DISCUSS] KIP-132: Augment KStream.print to allow extra parameters in the printed string

[Matthias J. Sax]

As there was no response from Marc, and we got KIP-160 as duplicate, I
am "closing" this KIP without a vote. If there are any objections please
let me know.


-Matthias

On 6/7/17 5:24 PM, Matthias J. Sax wrote:
> Hi,
> 
> as it turns out, we got two KIPs for the same Jira. KIP-160 duplicates
> KIP-132.
> 
> @Marc: as you did start the first KIP-132 but did not carry it on from
> some point on, I am wondering if you are still interested to work on it?
> 
> Sorry for this mix up. Hope we can resolve this properly...
> 
> 
> 
> -Matthias
> 
> 
> 
> On 4/6/17 8:46 PM, Matthias J. Sax wrote:
>> Hi Marc,
>>
>> any update on this KIP?
>>
>>
>> -Matthias
>>
>>
>> On 3/20/17 3:02 PM, Eno Thereska wrote:
>>> Hi Marc,
>>>
>>> Could you add more information in the motivation of the KIP as to what 
>>> problems this would solve? I can see how it can be done, but I don't yet 
>>> grok why it's useful. The KIP should contain more pain points/problems and 
>>> pose this as a solution. I know it's a small modification, but it's still 
>>> important to have a good motivation IMO.
>>>
>>> Thanks
>>> Eno
>>>
 On 20 Mar 2017, at 18:25, Matthias J. Sax  wrote:

 Sound reasonable Damian, but I guess, that's more a PR than KIP discussion.

 @Marc, I guess you can start a VOTE thread if there is no further feedback.


 -Matthias

 On 3/20/17 7:06 AM, Damian Guy wrote:
> Hi Marc,
>
> Thanks for the KIP. It mostly looks good to me. The only thing i'd change
> is using a null argument to use a default mapping. IMO it would be better
> if the existing print() method delegates to the new one supplying a
> KeyValueMapper that does the right thing.
>
> Thanks,
> Damian
>
> On Sat, 18 Mar 2017 at 14:25 Marc Juchli  wrote:
>
>> Thanks!
>>
>> I wanted to PING this thread. Not sure what the next steps of the KIP
>> process are?
>>
>> Kind regards,
>> Marc
>>
>> On Wed, Mar 15, 2017 at 9:13 PM Matthias J. Sax 
>> wrote:
>>
>>> Thanks for updating the KIP.
>>>
>>> It's in very good shape IMHO and I support this idea!
>>>
>>>
>>>
>>> -Matthias
>>>
>>>
>>> On 3/15/17 3:05 AM, Marc Juchli wrote:
 Dear Matthias,

 The KIP is updated. I think it now contains all the information on that
 page.

 Marc

 On Mon, Mar 13, 2017 at 9:37 PM Matthias J. Sax >>
 wrote:

> Marc,
>
> Thanks for the KIP.
>
> Can you please update the KIP in a way such that it is self contained.
> Right now, you link to all kind of other places making it hard to read
> the KIP.
>
> The KIP should be the "center of truth" -- if there is important
> information elsewhere, please c&p it into the KIP.
>
>
> Thanks a lot!
>
>
> -Matthias
>
>
>
> On 3/13/17 1:30 PM, Matthias J. Sax wrote:
>> Can you please add the KIP to this table:
>>
>>
>
>>>
>> https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals#KafkaImprovementProposals-KIPsunderdiscussion
>>
>> Thanks,
>>
>> Matthias
>>
>>
>> On 3/13/17 8:08 AM, Marc Juchli wrote:
>>> Dear all,
>>>
>>> The following describes KIP-132, which I just created. See:
>>>
>
>>>
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-132+-+Augment+KStream.print+to+allow+extra+parameters+in+the+printed+string
>>>
>>> Motivation
>>>
>>> As for now, KStream#print leads to a predefined output where key and
> value are
>>> printed with comma separation.
>>> KAFKA-4830 
>>> suggests
> to
>>> extend print in a way that it takes KeyValueMapper as a parameter.
>>> This will allow a user to change outputs according to the users
>>> demand.
>>> Public Interfaces
>>>
>>> The affected interface is KStream, which needs to be extended with
> another
>>> overloaded version of print:
>>>
>>> void print(final Serde keySerde,
>>>   final Serde valSerde,
>>>   final String streamName,
>>>   final KeyValueMapper mapper);
>>>
>>> Proposed Changes
>>>
>>> See pull request GH-2669 >> .
>>> This PR contains a discussion regarding KAFKA-4830
>>>  as well as
> KAFKA-4772
>>> .
>>>
>>> Compatibility, Deprecation, and Migration Plan
>>>

[jira] [Updated] (KAFKA-5354) MirrorMaker not preserving headers

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5354:
---
Status: Patch Available  (was: Open)

> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5354) MirrorMaker not preserving headers

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5354:
---
Fix Version/s: 0.11.0.0

> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
> Fix For: 0.11.0.0
>
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-5354) MirrorMaker not preserving headers

[Ismael Juma (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ismael Juma updated KAFKA-5354:
---
Priority: Blocker  (was: Major)

> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>Priority: Blocker
> Fix For: 0.11.0.0
>
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Work started] (KAFKA-4868) Optimize RocksDb config for fast recovery/bulk load

[Bill Bejeck (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4868?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Work on KAFKA-4868 started by Bill Bejeck.
--
> Optimize RocksDb config for fast recovery/bulk load
> ---
>
> Key: KAFKA-4868
> URL: https://issues.apache.org/jira/browse/KAFKA-4868
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
>Assignee: Bill Bejeck
> Fix For: 0.11.1.0
>
>
> RocksDb can be tuned to bulk-load data fast. Kafka Streams bulk-loads records 
> during recovery. It is likely we can use a different config to make recovery 
> faster, then revert to another config for normal operations like put/get. See 
> https://github.com/facebook/rocksdb/wiki/performance-benchmarks for examples. 
> Would be good to measure the performance gain as part of addressing this JIRA.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Work started] (KAFKA-5363) Add ability to batch restore and receive restoration stats.

[Bill Bejeck (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5363?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Work on KAFKA-5363 started by Bill Bejeck.
--
> Add ability to batch restore and receive restoration stats.
> ---
>
> Key: KAFKA-5363
> URL: https://issues.apache.org/jira/browse/KAFKA-5363
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Bill Bejeck
>Assignee: Bill Bejeck
>  Labels: kip
> Fix For: 0.11.1.0
>
>
> Currently, when restoring a state store in a Kafka Streams application, we 
> put one key-value at a time into the store.  
> This task aims to make this recovery more efficient by creating a new 
> interface with "restoreAll" functionality allowing for bulk writes by the 
> underlying state store implementation.  
> The proposal will also add "beginRestore" and "endRestore" callback methods 
> potentially used for 
> Tracking when the bulk restoration process begins and ends.
> Keeping track of the number of records and last offset restored.
> KIP: 
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-167%3A+Add+interface+for+the+state+store+restoration+process



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Comment Edited] (KAFKA-5354) MirrorMaker not preserving headers

[Michael Andre Pearce (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048320#comment-16048320
 ] 

Michael Andre Pearce edited comment on KAFKA-5354 at 6/13/17 7:58 PM:
--

[~junrao] yes it was a bug, I've created a test case, and also applied a fix, 
and raised a PR for trunk and 0.11 branch. If someone can review and merge? 
[~ijuma] this is a small bug fix for 0.11 if it could make it for the next RC1?

[~junrao] Thanks for spotting, sorry again for slow response.

btw, it seems i cannot assign the bug to myself


was (Author: michael.andre.pearce):
[~junrao] yes it was a bug, I've created a test case, and also applied a fix, 
and raised a PR for trunk and 0.11 branch. If someone can review and merge? 
[~ijuma] this is a small bug fix for 0.11 if it could make it for the next RC1?

[~junrao] Thanks for spotting, sorry again for slow response.

> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Comment Edited] (KAFKA-5354) MirrorMaker not preserving headers

[Michael Andre Pearce (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048320#comment-16048320
 ] 

Michael Andre Pearce edited comment on KAFKA-5354 at 6/13/17 7:58 PM:
--

[~junrao] yes it was a bug/issue, I've created a test case, and also applied a 
fix, and raised a PR for trunk and 0.11 branch. If someone can review and 
merge? [~ijuma] this is a small bug fix for 0.11 if it could make it for the 
next RC1?

[~junrao] Thanks for spotting, sorry again for slow response.

btw, it seems i cannot assign the bug to myself


was (Author: michael.andre.pearce):
[~junrao] yes it was a bug, I've created a test case, and also applied a fix, 
and raised a PR for trunk and 0.11 branch. If someone can review and merge? 
[~ijuma] this is a small bug fix for 0.11 if it could make it for the next RC1?

[~junrao] Thanks for spotting, sorry again for slow response.

btw, it seems i cannot assign the bug to myself

> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Comment Edited] (KAFKA-5354) MirrorMaker not preserving headers

[Michael Andre Pearce (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048320#comment-16048320
 ] 

Michael Andre Pearce edited comment on KAFKA-5354 at 6/13/17 7:57 PM:
--

[~junrao] yes it was a bug, I've created a test case, and also applied a fix, 
and raised a PR for trunk and 0.11 branch. If someone can review and merge? 
[~ijuma] this is a small bug fix for 0.11 if it could make it for the next RC1?

[~junrao] Thanks for spotting, sorry again for slow response.


was (Author: michael.andre.pearce):
[~junrao] yes it was a bug, I've created a test case, and also applied a fix, 
and raised a PR for trunk and 0.11 branch. If someone can review and merge?

Thanks for spotting, sorry again for slow response.

> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5354) MirrorMaker not preserving headers

[Michael Andre Pearce (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048320#comment-16048320
 ] 

Michael Andre Pearce commented on KAFKA-5354:
-

[~junrao] yes it was a bug, I've created a test case, and also applied a fix, 
and raised a PR for trunk and 0.11 branch. If someone can review and merge?

Thanks for spotting, sorry again for slow response.

> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Commented] (KAFKA-5354) MirrorMaker not preserving headers

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048318#comment-16048318
 ] 

ASF GitHub Bot commented on KAFKA-5354:
---

GitHub user michaelandrepearce opened a pull request:

https://github.com/apache/kafka/pull/3323

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/IG-Group/kafka KAFKA-5354-0.11.0

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3323.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3323


commit 8dc160a74b2cc660f073d56ed77587e545951573
Author: Michael Andre Pearce 
Date:   2017-06-13T19:52:50Z

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers




> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3323: KAFKA-5354: MirrorMaker not preserving headers

[michaelandrepearce]

GitHub user michaelandrepearce opened a pull request:

https://github.com/apache/kafka/pull/3323

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/IG-Group/kafka KAFKA-5354-0.11.0

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3323.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3323


commit 8dc160a74b2cc660f073d56ed77587e545951573
Author: Michael Andre Pearce 
Date:   2017-06-13T19:52:50Z

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-5354) MirrorMaker not preserving headers

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5354?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048298#comment-16048298
 ] 

ASF GitHub Bot commented on KAFKA-5354:
---

GitHub user michaelandrepearce opened a pull request:

https://github.com/apache/kafka/pull/3322

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/IG-Group/kafka KAFKA-5354

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3322.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3322


commit 9483f4590efa2c1e2662244a34ab046ca24e06c5
Author: Michael Andre Pearce 
Date:   2017-06-13T19:36:19Z

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers




> MirrorMaker not preserving headers
> --
>
> Key: KAFKA-5354
> URL: https://issues.apache.org/jira/browse/KAFKA-5354
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.11.0.0
>Reporter: Jun Rao
>
> Currently, it doesn't seem that MirrorMaker preserves headers in 
> BaseConsumerRecord. So, headers won't be preserved during mirroring.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3322: KAFKA-5354: MirrorMaker not preserving headers

[michaelandrepearce]

GitHub user michaelandrepearce opened a pull request:

https://github.com/apache/kafka/pull/3322

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/IG-Group/kafka KAFKA-5354

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3322.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3322


commit 9483f4590efa2c1e2662244a34ab046ca24e06c5
Author: Michael Andre Pearce 
Date:   2017-06-13T19:36:19Z

KAFKA-5354: MirrorMaker not preserving headers

Add test case to ensure fix and avoid regression
Update mirror maker for new consumer to preserve headers




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[GitHub] kafka pull request #3321: HOTFIX: Handle Connector version returning 'null' ...

[kkonstantine]

GitHub user kkonstantine opened a pull request:

https://github.com/apache/kafka/pull/3321

HOTFIX: Handle Connector version returning 'null' during plugin loading.



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/kkonstantine/kafka 
HOTFIX-Handle-null-version-returned-from-Connector-interface-during-plugin-loading

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3321.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3321


commit 72dc8982ae755cda5ba463344fe0857962e4914a
Author: Konstantine Karantasis 
Date:   2017-06-13T19:02:32Z

HOTFIX: Handle Connector version returning 'null' during plugin loading.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-5370) Replace uses of old consumer with the new consumer

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-5370?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16048236#comment-16048236
 ] 

ASF GitHub Bot commented on KAFKA-5370:
---

GitHub user vahidhashemian opened a pull request:

https://github.com/apache/kafka/pull/3320

KAFKA-5370 (WIP): Replace uses of the old consumer with the new consumer 
when possible

Also, methods in `ClientUtils` that are called by server or tools code 
should be introduced in `AdminUtils` with the implementation living in 
`AdminUtils`. All the existing callers apart from the old clients should call 
the `AdminUtils` methods.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/vahidhashemian/kafka KAFKA-5370

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3320.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3320


commit 8e09061514d7b4f30abe59dd9321c884d4aa9101
Author: Vahid Hashemian 
Date:   2017-06-13T15:36:31Z

KAFKA-5370: Replace use of old consumer with the new consumer when possible

- Uses of the old consumers in tools and tests where the new consumer would 
work as well (or better).
- Methods in `ClientUtils` that are called by server or tools code should 
be introduced in `AdminUtils` with the implementation living in `AdminUtils`. 
All the existing callers apart from the old clients should call the 
`AdminUtils` methods.




> Replace uses of old consumer with the new consumer 
> ---
>
> Key: KAFKA-5370
> URL: https://issues.apache.org/jira/browse/KAFKA-5370
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Vahid Hashemian
>Assignee: Vahid Hashemian
>Priority: Minor
>
> Where possible, use the new consumer In tools and tests instead of the old 
> consumer, and remove the deprecation warning.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3320: KAFKA-5370 (WIP): Replace uses of the old consumer...

[vahidhashemian]

GitHub user vahidhashemian opened a pull request:

https://github.com/apache/kafka/pull/3320

KAFKA-5370 (WIP): Replace uses of the old consumer with the new consumer 
when possible

Also, methods in `ClientUtils` that are called by server or tools code 
should be introduced in `AdminUtils` with the implementation living in 
`AdminUtils`. All the existing callers apart from the old clients should call 
the `AdminUtils` methods.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/vahidhashemian/kafka KAFKA-5370

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3320.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3320


commit 8e09061514d7b4f30abe59dd9321c884d4aa9101
Author: Vahid Hashemian 
Date:   2017-06-13T15:36:31Z

KAFKA-5370: Replace use of old consumer with the new consumer when possible

- Uses of the old consumers in tools and tests where the new consumer would 
work as well (or better).
- Methods in `ClientUtils` that are called by server or tools code should 
be introduced in `AdminUtils` with the implementation living in `AdminUtils`. 
All the existing callers apart from the old clients should call the 
`AdminUtils` methods.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-4653) Improve test coverage of RocksDBStore

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4653?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-4653:
---
Description: 
{{putAll}} - not covered
{{putInternal}} - exceptions



  was:
{{putAll}} - not covered
{{putInternal} - exceptions




> Improve test coverage of RocksDBStore
> -
>
> Key: KAFKA-4653
> URL: https://issues.apache.org/jira/browse/KAFKA-4653
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Jeyhun Karimov
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> {{putAll}} - not covered
> {{putInternal}} - exceptions



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-4653) Improve test coverage of RocksDBStore

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4653?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-4653:
---
Status: Patch Available  (was: Open)

> Improve test coverage of RocksDBStore
> -
>
> Key: KAFKA-4653
> URL: https://issues.apache.org/jira/browse/KAFKA-4653
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Jeyhun Karimov
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> {{putAll}} - not covered
> {{putInternal}} - exceptions



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-4658) Improve test coverage InMemoryKeyValueLoggedStore

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4658?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-4658:
---
Status: Patch Available  (was: Open)

> Improve test coverage InMemoryKeyValueLoggedStore
> -
>
> Key: KAFKA-4658
> URL: https://issues.apache.org/jira/browse/KAFKA-4658
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Jeyhun Karimov
> Fix For: 0.11.1.0
>
>
> {{putAll} not covered



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-4658) Improve test coverage InMemoryKeyValueLoggedStore

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4658?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-4658:
---
Description: {{putAll}} not covered  (was: {{putAll} not covered)

> Improve test coverage InMemoryKeyValueLoggedStore
> -
>
> Key: KAFKA-4658
> URL: https://issues.apache.org/jira/browse/KAFKA-4658
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Jeyhun Karimov
> Fix For: 0.11.1.0
>
>
> {{putAll}} not covered



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3319: MINOR: Verify mocks in all WorkerTest tests and do...

[ewencp]

GitHub user ewencp opened a pull request:

https://github.com/apache/kafka/pull/3319

MINOR: Verify mocks in all WorkerTest tests and don't unnecessarily 
mockStatic the Plugins class.



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/ewencp/kafka minor-worker-test-cleanup

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3319.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3319


commit 99b400a962c59f4186cfe406d38a5134ebaad2c6
Author: Ewen Cheslack-Postava 
Date:   2017-06-13T18:31:50Z

MINOR: Verify mocks in all WorkerTest tests and don't unnecessarily 
mockStatic the Plugins class.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-4656) Improve test coverage of CompositeReadOnlyKeyValueStore

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4656?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-4656:
---
Status: Patch Available  (was: Open)

> Improve test coverage of CompositeReadOnlyKeyValueStore
> ---
>
> Key: KAFKA-4656
> URL: https://issues.apache.org/jira/browse/KAFKA-4656
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Jeyhun Karimov
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> exceptions not covered



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-4659) Improve test coverage of CachingKeyValueStore

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4659?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-4659:
---
Status: Patch Available  (was: Open)

> Improve test coverage of CachingKeyValueStore
> -
>
> Key: KAFKA-4659
> URL: https://issues.apache.org/jira/browse/KAFKA-4659
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Jeyhun Karimov
>Priority: Minor
> Fix For: 0.11.1.0
>
>
> {{putIfAbsent}} mostly not covered



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Updated] (KAFKA-4655) Improve test coverage of CompositeReadOnlySessionStore

[Matthias J. Sax (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-4655?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Matthias J. Sax updated KAFKA-4655:
---
Status: Patch Available  (was: Open)

> Improve test coverage of CompositeReadOnlySessionStore
> --
>
> Key: KAFKA-4655
> URL: https://issues.apache.org/jira/browse/KAFKA-4655
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Damian Guy
>Assignee: Jeyhun Karimov
> Fix For: 0.11.1.0
>
>
> exceptions in fetch and internal iterator



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #3317: MINOR: Clean up in DeleteConsumerGroupTest even if...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3317


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Updated] (KAFKA-5382) Log recovery can fail if topic names contain one of the index suffixes

[Ewen Cheslack-Postava (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-5382?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Ewen Cheslack-Postava updated KAFKA-5382:
-
Fix Version/s: (was: 0.11.0.0.)
   0.11.0.0

> Log recovery can fail if topic names contain one of the index suffixes
> --
>
> Key: KAFKA-5382
> URL: https://issues.apache.org/jira/browse/KAFKA-5382
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.10.2.0, 0.10.2.1
>Reporter: Jason Gustafson
> Fix For: 0.11.0.0
>
>
> Our log recovery logic fails in 0.10.2 and prior releases if the topic name 
> contains "index" or "timeindex." The issue is this snippet:
> {code}
> val logFile =
>   if (filename.endsWith(TimeIndexFileSuffix))
> new File(file.getAbsolutePath.replace(TimeIndexFileSuffix, 
> LogFileSuffix))
>   else
> new File(file.getAbsolutePath.replace(IndexFileSuffix, 
> LogFileSuffix))
> if(!logFile.exists) {
>   warn("Found an orphaned index file, %s, with no corresponding log 
> file.".format(file.getAbsolutePath))
>   file.delete()
> }
> {code}
> The {{replace}} is a global replace, so the substituted filename is incorrect 
> if the topic contains the index suffix.
> Note this is already fixed in trunk and 0.11.0.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Re: [DISCUSS] KIP-150 - Kafka-Streams Cogroup

[Matthias J. Sax]

Hi,

I am just catching up on this thread. (1) as most people agree, we
should not add anything to KStreamBuilder (btw: we actually plan to move
#merge() to KStream and deprecate it on KStreamBuilder as it's a quite
unnatural API atm).

About specifying Serdes: there is still the idea to improve to overall
API from the current "we are adding more overloads"-pattern to a
builder-like pattern. This might make the whole discussion void if we do
this. Thus, it might make sense to keep this in mind (or even delay this
KIP?). It seems a waste of time to discuss all this if we are going to
chance it in 2 month anyway... Just saying.


-Matthias

On 6/13/17 8:05 AM, Michal Borowiecki wrote:
> You're right, I haven't thought of that.
> 
> Cheers,
> 
> Michał
> 
> 
> On 13/06/17 13:00, Kyle Winkelman wrote:
>> First, I would prefer not calling it aggregate because there are already
>> plenty of aggregate methods.
>>
>> Second, I dont think this would really work because after each aggregate
>> you now have a unique KTable (someone may want a table with 4 streams and
>> reuse those 4 in another table but with one more stream added) and unless
>> we completely duplicate everything every time this isnt really possible.
>> Additionally, the cogroup way just requires 1 more call to create two
>> different tables (normal, windowed, and session windowed) this new way
>> would require copying the aggregate chain.
>>
>> Another way to think about it is with cogroup we know that when they call
>> aggregate they arent going to be adding any more aggregators to that table
>> but your way requires us to assume they are done adding aggregators after
>> each call so we must return a ktable just to possibly not need to have
>> created it.
>>
>> On Jun 13, 2017 5:20 AM, "Michal Borowiecki" 
>> wrote:
>>
>>> Actually, just had a thought. It started with naming.
>>>
>>> Are we actually co-grouping these streams or are we co-aggregating them?
>>>
>>> After all, in each of the cogroup calls we are providing an Aggregator
>>> implementation.
>>>
>>>
>>> If they are really co-aggregated, why don't we turn this around:
>>> KGroupedStream grouped1 = builder.stream("topic1").groupByKey();
>>> KGroupedStream grouped2 = builder.stream("topic2").groupByKey();
>>> KGroupedStream grouped3 = builder.stream("topic3").groupByKey();
>>>
>>> KTable coagg = grouped1.aggregate(initializer1, aggregator1,
>>> aggValueSerde1) // this is the unchanged aggregate method
>>> .aggregate(grouped2, aggregator2)  // this is a new method
>>> .aggregate(grouped3, aggregator3); // ditto
>>>
>>> This means instead of adding cogroup methods on KGroupStream interface,
>>> adding aggregate method on KTable interface.
>>>
>>> Is that feasible?
>>>
>>> Cheers,
>>> Michał
>>>
>>> On 13/06/17 10:56, Michal Borowiecki wrote:
>>>
>>> Also, I still feel that putting initializer on the first cogroup can
>>> mislead users into thinking the first stream is in some way special.
>>> Just my 5c.
>>> Michał
>>>
>>> On 13/06/17 09:54, Michal Borowiecki wrote:
>>>
>>> Agree completely with the argument for serdes belonging in the same place
>>> as the state store name, which is in the aggregate method.
>>>
>>> Cheers,
>>>
>>> Michał
>>>
>>> On 12/06/17 18:20, Xavier Léauté wrote:
>>>
>>> I think we are discussing two separate things here, so it might be worth
>>> clarifying:
>>>
>>> 1) the position of the initializer with respect to the aggregators. If I
>>> understand correctly, Guozhang seems to think it is more natural to specify
>>> the initializer first, despite it not bearing any relation to the first
>>> aggregator. I can see the argument for specifying the initializer first,
>>> but I think it is debatable whether mixing it into the first cogroup call
>>> leads to a cleaner API or not.
>>>
>>> 2) where the serde should be defined (if necessary). Looking at our
>>> existing APIs in KGroupedStreams, we always offer two aggregate()
>>> methods. The first one takes the name of the store and associated aggregate
>>> value serde e.g. KGroupedStream.aggregate(Initializer initializer,
>>> Aggregator aggregator, Serde aggValueSerde,
>>> String queryableStoreName)
>>> The second one only takes a state store supplier, and does not specify any
>>> serde, e.g. KGroupedStream.aggregate(Initializer
>>> initializer, Aggregator aggregator, final
>>> StateStoreSupplier storeSupplier)
>>> Presumably, when specifying a state store supplier it shouldn't be
>>> necessary to specify an aggregate value serde, since the provided
>>> statestore might not need to serialize the values (e.g. it may just keep
>>> them as regular objects in heap) or it may have its own
>>> internal serialization format.
>>>
>>> For consistency I think it would be valuable to preserve the same two
>>> aggregate methods for cogroup as well. Since the serde is only required in
>>> one of the two cases, I believe the serde has no place in the first
>>> cogroup() call and should only have to be specified as part of

Re: [DISCUSS]: KIP-159: Introducing Rich functions to Streams

[Matthias J. Sax]

Hi,

I would like to push this discussion further. It seems we got nice
alternatives (thanks for the summary Jeyhun!).

With respect to RichFunctions and allowing them to be stateful, I have
my doubt as expressed already. From my understanding, the idea was to
give access to record metadata information only. If you want to do a
stateful computation you should rather use #transform().

Furthermore, as pointed out, we would need to switch to a
supplier-pattern introducing many more overloads.

For those reason, I advocate for a simple interface with a single method
that passes in a RecordContext object.


-Matthias


On 6/6/17 5:15 PM, Guozhang Wang wrote:
> Thanks for the comprehensive summary!
> 
> Personally I'd prefer the option of passing RecordContext as an additional
> parameter into he overloaded function. But I'm also open to other arguments
> if there are sth. that I have overlooked.
> 
> Guozhang
> 
> 
> On Mon, Jun 5, 2017 at 3:19 PM, Jeyhun Karimov  wrote:
> 
>> Hi,
>>
>> Thanks for your comments Matthias and Guozhang.
>>
>> Below I mention the quick summary of the main alternatives we looked at to
>> introduce the Rich functions (I will refer to it as Rich functions until we
>> find better/another name). Initially the proposed alternatives was not
>> backwards-compatible, so I will not mention them.
>> The related discussions are spread in KIP-149 and in this KIP (KIP-159)
>> discussion threads.
>>
>>
>>
>> 1. The idea of rich functions came into the stage with KIP-149, in
>> discussion thread. As a result we extended KIP-149 to support Rich
>> functions as well.
>>
>> 2.  To as part of the Rich functions, we provided init (ProcessorContext)
>> method. Afterwards, Dammian suggested that we should not provide
>> ProcessorContext to users. As a result, we separated the two problems into
>> two separate KIPs, as it seems they can be solved in parallel.
>>
>> - One approach we considered was :
>>
>> public interface ValueMapperWithKey {
>> VR apply(final K key, final V value);
>> }
>>
>> public interface RichValueMapper extends RichFunction{
>> }
>>
>> public interface RichFunction {
>> void init(RecordContext recordContext);
>> void close();
>> }
>>
>> public interface RecordContext {
>> String applicationId();
>> TaskId taskId();
>> StreamsMetrics metrics();
>> String topic();
>> int partition();
>> long offset();
>> long timestamp();
>> Map appConfigs();
>> Map appConfigsWithPrefix(String prefix);
>> }
>>
>>
>> public interface ProcessorContext extends RecordContext {
>>// all methods but the ones in RecordContext
>> }
>>
>> As a result:
>> * . All "withKey" and "withoutKey" interfaces can be converted to their
>> Rich counterparts (with empty init() and close() methods)
>> *. All related Processors will accept Rich interfaces in their
>> constructors.
>> *. So, we convert the related "withKey" or "withoutKey" interfaces to Rich
>> interface while building the topology and initialize the related processors
>> with Rich interfaces only.
>> *. We will not need to overloaded methods for rich functions as Rich
>> interfaces extend withKey interfaces. We will just check the object type
>> and act accordingly.
>>
>>
>>
>>
>> 3. There was some thoughts that the above approach does not support lambdas
>> so we should support only one method, only init(RecordContext), as part of
>> Rich interfaces.
>> This is still in discussion. Personally I think Rich interfaces are by
>> definition lambda-free and we should not care much about it.
>>
>>
>> 4. Thanks to Matthias's discussion, an alternative we considered was to
>> pass in the RecordContext as method parameter.  This might even allow to
>> use Lambdas and we could keep the name RichFunction as we preserve the
>> nature of being a function.
>> "If you go with `init()` and `close()` we basically
>> allow users to have an in-memory state for a function. Thus, we cannot
>> share a single instance of RichValueMapper (etc) over multiple tasks and
>> we would need a supplier pattern similar to #transform(). And this would
>> "break the flow" of the API, as (Rich)ValueMapperSupplier would not
>> inherit from ValueMapper and thus we would need many new overload for
>> KStream/KTable classes". (Copy paste from Matthias's email)
>>
>>
>> Cheers,
>> Jeyhun
>>
>>
>> On Mon, Jun 5, 2017 at 5:18 AM Matthias J. Sax 
>> wrote:
>>
>>> Yes, we did consider this, and there is no consensus yet what the best
>>> alternative is.
>>>
>>> @Jeyhun: the email thread got pretty long. Maybe you can give a quick
>>> summary of the current state of the discussion?
>>>
>>>
>>> -Matthias
>>>
>>> On 6/4/17 6:04 PM, Guozhang Wang wrote:
 Thanks for the explanation Jeyhun and Matthias.

 I have just read through both KIP-149 and KIP-159 and am wondering if
>> you
 guys have considered a slight different approach for rich function,
>> that
>>> is
 to add the `RecordContext` into the apply functions as an additional
>>>

[GitHub] kafka pull request #3318: HOTFIX: Fix invalid long format conversion in requ...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/3318


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [DISCUSS]: KIP-149: Enabling key access in ValueTransformer, ValueMapper, and ValueJoiner

[Matthias J. Sax]

I agree with Guozhang's second point. This change does not seem backward
compatible.

As we don't have to support lambdas, it might be the easiest thing to
just extend the current interface:

> public interface ValueTransformerWithKey extends 
> ValueTransformer 

When plugging the topology together, we can check if we get the
`withKey` variant and use a corresponding runtime class for execution,
so we get only a single time check. Thus, for the `withKey` variant, the
will be a `transfrom(V value)` method, but we will never call it.

Maybe we could make `ValueTransformerWithKey` an abstract class with a
`final` no-op implemenation of `transform(V value)` ?


-Matthias


On 6/6/17 4:58 PM, Guozhang Wang wrote:
> Jeyhun, Matthias:
> 
> Thanks for the explanation, I overlooked the repartition argument
> previously.
> 
> 1) Based on that argument I'm convinced of having ValueMapperWithKey /
> ValueJoinerWithKey / ValueTransformerWithKey; though I'm still not
> convinced with ReducerWithKey and InitializerWithKey since for the former
> it can be covered with `aggregate` completely and with latter I have seen
> little use cases with it.
> 
> 2) Another comment is on public interface ValueTransformer extends
> ValueTransformerCommon:
> 
> I think changing the interface to extend from a new interface is not binary
> compatible though source compatible, i.e. users still need to recompile
> their code though no need to make code changes. We may need to mention that
> in the upgrade path if we want to keep it that way.
> 
> Guozhang
> 
> On Mon, Jun 5, 2017 at 2:28 PM, Jeyhun Karimov  wrote:
> 
>> Hi,
>>
>>
>> Sorry for late reply. Just to make everybody in sync, the current version
>> of KIP supports lambdas. "withKey" (ValueMapperWithKey) interfaces are
>> independent, meaning they do not extend from "withoutKey" (ValueMapper)
>> interfaces.
>>
>>
>> I agree with Guozhang, and I am personally a bit reluctant to increase
>> overloaded methods in public APIs but it seems this is only way to solve
>> all related jira issues.
>> However, the most overloaded methods will be with ValueJoiner type, which
>> will be with ValueJoinerWithKey with new overloaded methods. Other
>> interfaces require mostly 1 extra overload.
>>
>>
 I would suggest not doing it if user pop it up, but rather suggesting
>> them
 to use `map`
>>
>> I agree with Matthias as the core idea of this KIP was to collect all
>> related jira issues and propose one-shot solution for all. Afterwards, we
>> broke its scope into 2 KIPs (149 and 159).
>>
>> Cheers,
>> Jeyhun
>>
>>
>>
>> On Mon, Jun 5, 2017 at 7:55 AM Matthias J. Sax 
>> wrote:
>>
>>> I guess I missunderstood you. Your are right that overloading the method
>>> would also work. As both interfaces will be independent from each other,
>>> there should be no problem.
>>>
>>> The initial proposal was to use
>>>
 interface ValueMapperWithKey extends ValueMapper
>>>
>>> and this would break Lambda. We totally missed, that we don't need new
>>> methods but only only overlaods. Great you point this out! I was not
>>> quite happy with the newly added methods.
>>>
> I would suggest not doing it if user pop it up, but rather suggesting
>>> them
> to use `map`
>>>
>>> But this might introduce unnecessary re-partitioning for downstream
>>> operators. I don't thinks that a good suggestion. But if we only add new
>>> overloads for mapValue(), flatMapValue() etc, I don't see a big issue
>>> with "overstaffing" the API (what is btw a very valid concern).
>>>
>>>
>>> -Matthias
>>>
>>>
>>> On 6/4/17 10:37 PM, Guozhang Wang wrote:
 On Sun, Jun 4, 2017 at 8:41 PM, Matthias J. Sax >>
 wrote:

> We started with this proposal but it does not allow for Lambdas (in
>> case
> you want key access). Do you think preserving Lambdas is not important
> enough for this case -- I agree that there is some price to be paid to
> keep Lambdas.
>

 Not sure if I understand this comment.. My main point is not on
>> changing
 the proposal but just reduce it scope to `ValueJoinerWithKey` only; and
 even if we want to keep them all, I'd suggest we just implement the
>> added
 APIs by using the `KeyValueMapper` for `ValueMapperWithKeys`, etc,
>> which
 seems simpler to me. Does that affect lambda expression usage?

>
> About API consistency: I can't remember a concrete user request to
>> have
> key access in all operators. But I am pretty sure, if we only add it
>> for
> some, this request will pop up quickly. IMHO, it's better to do it all
> in one shot. (But I am not strict about it if most people think we
> should limit it to what was requested by users.)
>
>
 I would suggest not doing it if user pop it up, but rather suggesting
>>> them
 to use `map` etc directly but set the key unchanged rather than
>>> providing a
 new operator for it. To me some syntax sugars like this seems of less
 valuable than others (li

[GitHub] kafka pull request #3318: HOTFIX: Fix invalid long format conversion in requ...

[hachikuji]

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/3318

HOTFIX: Fix invalid long format conversion in request logger message



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka hotfix-request-metric-udpate

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/3318.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #3318


commit 659930d1dbda09690f9b046e096dee77a0318c83
Author: Jason Gustafson 
Date:   2017-06-13T17:22:57Z

HOTFIX: Fix invalid long format conversion in request logger message




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---