[jira] [Commented] (KAFKA-2284) ConsumerRebalanceListener receives wrong type in partitionOwnership values

[Bernard Leach (JIRA)]

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

Bernard Leach commented on KAFKA-2284:
--

Just came across this as a failing test in 
https://github.com/apache/kafka/pull/2101 where I had fixed the underlying bug. 
 I've separated the fix out to a separate PR 
https://github.com/apache/kafka/pull/2102.

> ConsumerRebalanceListener receives wrong type in partitionOwnership values
> --
>
> Key: KAFKA-2284
> URL: https://issues.apache.org/jira/browse/KAFKA-2284
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.8.2.0
>Reporter: E. Sammer
>Assignee: Neha Narkhede
>
> The ConsumerRebalanceListener's beforeReleasingPartitions() method is 
> supposed to receive an arg of Map (topic -> 
> Set(partitions)). Even though the type of the map value is specified as a 
> java.util.Set, a scala.collection.convert.Wrappers$JSetWrapper is passed 
> instead which does not implement Set causing a class cast exception as 
> soon as one attempts to access any value of the map. It looks as if this 
> method was never tested against the actual types specified by the interface.
> Here's what happens if you call {{Set foo = 
> partitionOwnership.get(topic)}}:
> {code}
> 2015-06-18 07:28:43,776 
> (search-consumer_esammer-mbp.local-1434637723383-12126c1b_watcher_executor) 
> [WARN - 
> com.rocana.search.consumer.ConsumerService$1.beforeReleasingPartitions(ConsumerService.java:246)]
>  Exception while rebalancing!
> java.lang.ClassCastException: scala.collection.convert.Wrappers$JSetWrapper 
> cannot be cast to java.util.Set
>   at 
> com.rocana.search.consumer.IndexConsumerWorker.onRebalance(IndexConsumerWorker.java:80)
>   at 
> com.rocana.search.consumer.ConsumerService$1.beforeReleasingPartitions(ConsumerService.java:244)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:675)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1$$anonfun$apply$mcV$sp$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:625)
>   at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:142)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcV$sp(ZookeeperConsumerConnector.scala:619)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:616)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:616)
>   at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:615)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:568)
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-2284) ConsumerRebalanceListener receives wrong type in partitionOwnership values

[ASF GitHub Bot (JIRA)]

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

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

GitHub user leachbj opened a pull request:

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

KAFKA-2284: corrects value type in beforeReleasingPartitions

Previously the values in the map were scala Set values, now they
are correctly java.util.Set as advertised by API.

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

$ git pull https://github.com/leachbj/kafka 
2284-java-set-beforeReleasingPartitions

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

https://github.com/apache/kafka/pull/2102.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 #2102


commit 3f249614da15cc9239b4aec55f633a1f146359ea
Author: Bernard Leach 
Date:   2016-11-04T01:05:15Z

KAFKA-2284: corrects value type in beforeReleasingPartitions

Previously the values in the map were scala Set values, now they
are correctly java.util.Set as advertised by API.




> ConsumerRebalanceListener receives wrong type in partitionOwnership values
> --
>
> Key: KAFKA-2284
> URL: https://issues.apache.org/jira/browse/KAFKA-2284
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.8.2.0
>Reporter: E. Sammer
>Assignee: Neha Narkhede
>
> The ConsumerRebalanceListener's beforeReleasingPartitions() method is 
> supposed to receive an arg of Map (topic -> 
> Set(partitions)). Even though the type of the map value is specified as a 
> java.util.Set, a scala.collection.convert.Wrappers$JSetWrapper is passed 
> instead which does not implement Set causing a class cast exception as 
> soon as one attempts to access any value of the map. It looks as if this 
> method was never tested against the actual types specified by the interface.
> Here's what happens if you call {{Set foo = 
> partitionOwnership.get(topic)}}:
> {code}
> 2015-06-18 07:28:43,776 
> (search-consumer_esammer-mbp.local-1434637723383-12126c1b_watcher_executor) 
> [WARN - 
> com.rocana.search.consumer.ConsumerService$1.beforeReleasingPartitions(ConsumerService.java:246)]
>  Exception while rebalancing!
> java.lang.ClassCastException: scala.collection.convert.Wrappers$JSetWrapper 
> cannot be cast to java.util.Set
>   at 
> com.rocana.search.consumer.IndexConsumerWorker.onRebalance(IndexConsumerWorker.java:80)
>   at 
> com.rocana.search.consumer.ConsumerService$1.beforeReleasingPartitions(ConsumerService.java:244)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.kafka$consumer$ZookeeperConsumerConnector$ZKRebalancerListener$$rebalance(ZookeeperConsumerConnector.scala:675)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1$$anonfun$apply$mcV$sp$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:625)
>   at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:142)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcV$sp(ZookeeperConsumerConnector.scala:619)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:616)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply(ZookeeperConsumerConnector.scala:616)
>   at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:615)
>   at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:568)
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2102: KAFKA-2284: corrects value type in beforeReleasing...

[leachbj]

GitHub user leachbj opened a pull request:

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

KAFKA-2284: corrects value type in beforeReleasingPartitions

Previously the values in the map were scala Set values, now they
are correctly java.util.Set as advertised by API.

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

$ git pull https://github.com/leachbj/kafka 
2284-java-set-beforeReleasingPartitions

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

https://github.com/apache/kafka/pull/2102.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 #2102


commit 3f249614da15cc9239b4aec55f633a1f146359ea
Author: Bernard Leach 
Date:   2016-11-04T01:05:15Z

KAFKA-2284: corrects value type in beforeReleasingPartitions

Previously the values in the map were scala Set values, now they
are correctly java.util.Set as advertised by API.




---
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-4371) Sporadic ConnectException shuts down the whole connect process

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava commented on KAFKA-4371:
--

[~sagarrao] I'm a bit confused by the report. You say it takes down the whole 
connect process, but then talk about using the REST API after the failure. Does 
the process actually die or are you just saying that the process no longer 
continues to process data? If it's the former, then it'd be a bug with the 
framework. If it's a latter, this is just an issue with the JDBC connector 
which is why [~skyahead] was saying the bug probably should be filed in the 
JDBC connector's repository.


> Sporadic ConnectException shuts down the whole connect process
> --
>
> Key: KAFKA-4371
> URL: https://issues.apache.org/jira/browse/KAFKA-4371
> Project: Kafka
>  Issue Type: Bug
>Reporter: Sagar Rao
>Priority: Critical
>
> I had setup a 2 node distributed kafka-connect process. Everything went well 
> and I could see lot of data flowing into the relevant kafka topics.
> After some time, JDBCUtils.getCurrentTimeOnDB threw a ConnectException with 
> the following stacktrace:
> The last packet successfully received from the server was 792 milliseconds 
> ago.  The last packet sent successfully to the server was 286 milliseconds 
> ago. (io.confluent.connect.jdbc.source.JdbcSourceTask:234)
> [2016-11-02 12:42:06,116] ERROR Failed to get current time from DB using 
> query select CURRENT_TIMESTAMP; on database MySQL 
> (io.confluent.connect.jdbc.util.JdbcUtils:226)
> com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link 
> failure
> The last packet successfully received from the server was 1,855 milliseconds 
> ago.  The last packet sent successfully to the server was 557 milliseconds 
> ago.
>at sun.reflect.GeneratedConstructorAccessor51.newInstance(Unknown 
> Source)
>at 
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
>at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
>at com.mysql.jdbc.Util.handleNewInstance(Util.java:411)
>at 
> com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:1117)
>at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3829)
>at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2449)
>at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2629)
>at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2719)
>at 
> com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:2155)
>at 
> com.mysql.jdbc.PreparedStatement.execute(PreparedStatement.java:1379)
>at 
> com.mysql.jdbc.StatementImpl.createResultSetUsingServerFetch(StatementImpl.java:651)
>at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1527)
>at 
> io.confluent.connect.jdbc.util.JdbcUtils.getCurrentTimeOnDB(JdbcUtils.java:220)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.executeQuery(TimestampIncrementingTableQuerier.java:157)
>at 
> io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:78)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:57)
>at 
> io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:207)
>at 
> org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:155)
>at 
> org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
>at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
>at 
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
>at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
>at java.lang.Thread.run(Thread.java:745)
> Caused by: java.net.SocketException: Broken pipe (Write failed)
>at java.net.SocketOutputStream.socketWrite0(Native Method)
>at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:109)
>at java.net.SocketOutputStream.write(SocketOutputStream.java:153)
>at 
> java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
>at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
>at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3810)
>... 20 more
> This was just a minor glitch to the connection as the ec2 isntances are able 
> to connect to the Mysql Aurora instances without any issues.
> But, after 

[jira] [Commented] (KAFKA-4373) Kafka Consumer API jumping offsets

[huxi (JIRA)]

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

huxi commented on KAFKA-4373:
-

Could you paste your client code and the stack trace?

> Kafka Consumer API jumping offsets
> --
>
> Key: KAFKA-4373
> URL: https://issues.apache.org/jira/browse/KAFKA-4373
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Reporter: Srinivasan Venkatraman
>
> Hi,
> I am using Kafka Version 0.10.0.1 and java consumer API to consume messages 
> from a topic. We are using a single node kafka and zookeeper. It is sometimes 
> observed that the consumer is losing a bulk of message. We are unable to find 
> the exact reason to replicate the issue.
> The scenario is:
> Consumer polls the topic.
> Fetches the messages and gives it to a thread pool to handle the message.
> Waits for the threads to return and then commits the offsets.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4367) MirrorMaker shuts down gracefully without actually being stopped

[huxi (JIRA)]

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

huxi commented on KAFKA-4367:
-

Did you run it with nohup?

> MirrorMaker shuts down gracefully without actually being stopped
> 
>
> Key: KAFKA-4367
> URL: https://issues.apache.org/jira/browse/KAFKA-4367
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.9.0.1
> Environment: RHEL 7
>Reporter: Alex
>
> MirrorMaker stops working without being stopped. From 30 minutes to 20 hours. 
> No clue why this problem occurs.
> Start:
> bin/kafka-mirror-maker.sh --new.consumer --consumer.config 
> config/ssl_mirroring_consumer.properties --producer.config 
> config/ssl_mirroring_producer.properties --whitelist 
> "TOPIC1|TOPIC2|TOPIC3|TOPIC4" --num.streams 20 &> /dev/null &
> 
>   kafka-mirror-maker.log
> 
> [2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
> CEP.FS.IN-175 with base offset offset 15015 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
> CEP.FS.IN-151 with base offset offset 15066 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,003] TRACE Nodes with data ready to send: [Node(8, 
> 10.126.0.2, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,003] TRACE Created 1 produce requests: 
> [ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.producer.internals.Sender$1@483c4c7a, 
> request=RequestSend(header={api_key=0,api_version=1,correlation_id=219685,client_id=producer-1},
>  
> body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=133,record_set=java.nio.HeapByteBuffer[pos=0
>  lim=9085 cap=16384]}]}]}), createdTimeMs=1478017412003, sendTimeMs=0)] 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,008] TRACE Returning fetched records for assigned 
> partition CEP.FS.IN-172 and update consumed position to 3869316 
> (org.apache.kafka.clients.consumer.internals.Fetcher)
> [2016-11-01 19:23:32,008] TRACE [mirrormaker-thread-7] Sending message with 
> value size 485 and offset 3869315 (kafka.tools.MirrorMaker$MirrorMakerThread)
> [2016-11-01 19:23:32,008] TRACE Sending record 
> ProducerRecord(topic=CEP.FS.IN, partition=null, key=null, value=[B@12a54f5a 
> with callback kafka.tools.MirrorMaker$MirrorMakerProducerCallback@5ea65b8f to 
> topic CEP.FS.IN partition 160 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2016-11-01 19:23:32,008] TRACE Allocating a new 16384 byte message buffer 
> for topic CEP.FS.IN partition 160 
> (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2016-11-01 19:23:32,008] TRACE Waking up the sender since topic CEP.FS.IN 
> partition 160 is either full or getting a new batch 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2016-11-01 19:23:32,010] TRACE Received produce response from node 7 with 
> correlation id 219684 (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
> CEP.FS.IN-106 with base offset offset 15086 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
> CEP.FS.IN-124 with base offset offset 15095 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,010] TRACE Nodes with data ready to send: [Node(7, 
> 10.126.0.1, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] INFO Start clean shutdown. 
> (kafka.tools.MirrorMaker$)
> [2016-11-01 19:23:32,010] TRACE Created 1 produce requests: 
> [ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.producer.internals.Sender$1@44b788c7, 
> request=RequestSend(header={api_key=0,api_version=1,correlation_id=219686,client_id=producer-1},
>  
> body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=160,record_set=java.nio.HeapByteBuffer[pos=0
>  lim=511 cap=16384]}]}]}), createdTimeMs=1478017412010, sendTimeMs=0)] 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] INFO Shutting down consumer threads. 
> (kafka.tools.MirrorMaker$)
> [2016-11-01 19:23:32,011] INFO [mirrormaker-thread-0] mirrormaker-thread-0 
> shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
> [2016-11-01 19:23:32,011] INFO [mirrormaker-thread-1] mirrormaker-thread-1 
> shutting down 

[jira] [Issue Comment Deleted] (KAFKA-4375) Kafka consumer may swallow some interrupts meant for the calling thread

[JIRA]

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

Stig Rohde Døssing updated KAFKA-4375:
--
Comment: was deleted

(was: Reset thread interrupt state in a few places where InterruptedException 
is caught in threads that the Kafka code didn't start.)

> Kafka consumer may swallow some interrupts meant for the calling thread
> ---
>
> Key: KAFKA-4375
> URL: https://issues.apache.org/jira/browse/KAFKA-4375
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.1.0
>Reporter: Stig Rohde Døssing
>Assignee: Stig Rohde Døssing
>
> Apache Storm has added a new data source ("spout") based on the Kafka 0.9 
> consumer. Storm interacts with the consumer by having one thread per spout 
> instance loop calls to poll/commitSync etc. When Storm shuts down, another 
> thread indicates that the looping threads should shut down by interrupting 
> them, and joining them.
> If one of the looping threads happen to be interrupted while executing 
> certain sleeps in some consumer methods (commitSync and committed at least), 
> the interrupt can be lost because they contain a call to SystemTime.sleep, 
> which swallows the interrupt.
> Is this behavior by design, or can SystemTime be changed to reset the thread 
> interrupt flag when catching an InterruptedException? 
> I haven't checked the rest of the client code, so it's possible that this is 
> an issue in other parts of the code too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-4375) Kafka consumer may swallow some interrupts meant for the calling thread

[JIRA]

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

Stig Rohde Døssing updated KAFKA-4375:
--
Affects Version/s: 0.10.1.0
   Status: Patch Available  (was: Open)

Reset thread interrupt state in a few places where InterruptedException is 
caught in threads that the Kafka code didn't start.

> Kafka consumer may swallow some interrupts meant for the calling thread
> ---
>
> Key: KAFKA-4375
> URL: https://issues.apache.org/jira/browse/KAFKA-4375
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.1.0
>Reporter: Stig Rohde Døssing
>Assignee: Stig Rohde Døssing
>
> Apache Storm has added a new data source ("spout") based on the Kafka 0.9 
> consumer. Storm interacts with the consumer by having one thread per spout 
> instance loop calls to poll/commitSync etc. When Storm shuts down, another 
> thread indicates that the looping threads should shut down by interrupting 
> them, and joining them.
> If one of the looping threads happen to be interrupted while executing 
> certain sleeps in some consumer methods (commitSync and committed at least), 
> the interrupt can be lost because they contain a call to SystemTime.sleep, 
> which swallows the interrupt.
> Is this behavior by design, or can SystemTime be changed to reset the thread 
> interrupt flag when catching an InterruptedException? 
> I haven't checked the rest of the client code, so it's possible that this is 
> an issue in other parts of the code too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4376) Add scala 2.12 support

[Bernard Leach (JIRA)]

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

Bernard Leach commented on KAFKA-4376:
--

2.12 will need to be built using java8, currently the jenkins builds are java7 
only.

> Add scala 2.12 support
> --
>
> Key: KAFKA-4376
> URL: https://issues.apache.org/jira/browse/KAFKA-4376
> Project: Kafka
>  Issue Type: Improvement
>  Components: build
>Affects Versions: 0.10.1.0, 0.10.0.1
>Reporter: Bernard Leach
>
> Now that Scala 2.12 has now been officially released releasing 2.12 builds of 
> the kafka artifacts will allow downstream projects such as reactive-kafka via 
> scalatest-embedded-kafka to release 2.12 builds.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2101: KAFKA-4377: remove deprecated scala.collection.Jav...

[leachbj]

GitHub user leachbj opened a pull request:

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

KAFKA-4377: remove deprecated scala.collection.JavaConversions calls

JavaConversions are deprecated in 2.12 in favour of JavaConverters.

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

$ git pull https://github.com/leachbj/kafka 4377-java-converters

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

https://github.com/apache/kafka/pull/2101.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 #2101


commit ec37151abac171164bf2827d6b8ff37642f4
Author: Bernard Leach 
Date:   2016-11-02T01:45:45Z

KAFKA-4377: remove deprecated scala.collection.JavaConversions calls

JavaConversions are deprecated in 2.12 in favour of JavaConverters.




---
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] [Assigned] (KAFKA-4375) Kafka consumer may swallow some interrupts meant for the calling thread

[JIRA]

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

Stig Rohde Døssing reassigned KAFKA-4375:
-

Assignee: Stig Rohde Døssing

> Kafka consumer may swallow some interrupts meant for the calling thread
> ---
>
> Key: KAFKA-4375
> URL: https://issues.apache.org/jira/browse/KAFKA-4375
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Stig Rohde Døssing
>Assignee: Stig Rohde Døssing
>
> Apache Storm has added a new data source ("spout") based on the Kafka 0.9 
> consumer. Storm interacts with the consumer by having one thread per spout 
> instance loop calls to poll/commitSync etc. When Storm shuts down, another 
> thread indicates that the looping threads should shut down by interrupting 
> them, and joining them.
> If one of the looping threads happen to be interrupted while executing 
> certain sleeps in some consumer methods (commitSync and committed at least), 
> the interrupt can be lost because they contain a call to SystemTime.sleep, 
> which swallows the interrupt.
> Is this behavior by design, or can SystemTime be changed to reset the thread 
> interrupt flag when catching an InterruptedException? 
> I haven't checked the rest of the client code, so it's possible that this is 
> an issue in other parts of the code too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4377) Address 2.12 deprecations

[ASF GitHub Bot (JIRA)]

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

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

GitHub user leachbj opened a pull request:

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

KAFKA-4377: remove deprecated scala.collection.JavaConversions calls

JavaConversions are deprecated in 2.12 in favour of JavaConverters.

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

$ git pull https://github.com/leachbj/kafka 4377-java-converters

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

https://github.com/apache/kafka/pull/2101.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 #2101


commit ec37151abac171164bf2827d6b8ff37642f4
Author: Bernard Leach 
Date:   2016-11-02T01:45:45Z

KAFKA-4377: remove deprecated scala.collection.JavaConversions calls

JavaConversions are deprecated in 2.12 in favour of JavaConverters.




> Address 2.12 deprecations
> -
>
> Key: KAFKA-4377
> URL: https://issues.apache.org/jira/browse/KAFKA-4377
> Project: Kafka
>  Issue Type: Sub-task
>  Components: core
>Reporter: Bernard Leach
>
> The 2.12 compiler is generating a large number of deprecation warnings, those 
> that can be resolved in a 2.10 compatible way should be fixed.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2100: KAFKA-4375: Reset thread interrupted state in a fe...

[srdo]

GitHub user srdo opened a pull request:

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

KAFKA-4375: Reset thread interrupted state in a few places where Inte…

…rruptedException is caught in a thread Kafka didn't start.

See https://issues.apache.org/jira/browse/KAFKA-4375

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

$ git pull https://github.com/srdo/kafka KAFKA-4375

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

https://github.com/apache/kafka/pull/2100.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 #2100


commit f2fa7a09f0bb29b4c4245449714c9e0e2cda73be
Author: Stig Rohde Døssing 
Date:   2016-11-03T21:45:36Z

KAFKA-4375: Reset thread interrupted state in a few places where 
InterruptedException is caught in a thread Kafka didn't start




---
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-4375) Kafka consumer may swallow some interrupts meant for the calling thread

[ASF GitHub Bot (JIRA)]

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

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

GitHub user srdo opened a pull request:

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

KAFKA-4375: Reset thread interrupted state in a few places where Inte…

…rruptedException is caught in a thread Kafka didn't start.

See https://issues.apache.org/jira/browse/KAFKA-4375

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

$ git pull https://github.com/srdo/kafka KAFKA-4375

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

https://github.com/apache/kafka/pull/2100.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 #2100


commit f2fa7a09f0bb29b4c4245449714c9e0e2cda73be
Author: Stig Rohde Døssing 
Date:   2016-11-03T21:45:36Z

KAFKA-4375: Reset thread interrupted state in a few places where 
InterruptedException is caught in a thread Kafka didn't start




> Kafka consumer may swallow some interrupts meant for the calling thread
> ---
>
> Key: KAFKA-4375
> URL: https://issues.apache.org/jira/browse/KAFKA-4375
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Stig Rohde Døssing
>
> Apache Storm has added a new data source ("spout") based on the Kafka 0.9 
> consumer. Storm interacts with the consumer by having one thread per spout 
> instance loop calls to poll/commitSync etc. When Storm shuts down, another 
> thread indicates that the looping threads should shut down by interrupting 
> them, and joining them.
> If one of the looping threads happen to be interrupted while executing 
> certain sleeps in some consumer methods (commitSync and committed at least), 
> the interrupt can be lost because they contain a call to SystemTime.sleep, 
> which swallows the interrupt.
> Is this behavior by design, or can SystemTime be changed to reset the thread 
> interrupt flag when catching an InterruptedException? 
> I haven't checked the rest of the client code, so it's possible that this is 
> an issue in other parts of the code too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2099: MINOR: Fix re-raise of python error in system test...

[ewencp]

GitHub user ewencp opened a pull request:

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

MINOR: Fix re-raise of python error in system tests



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

$ git pull https://github.com/ewencp/kafka fix-python-reraise

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

https://github.com/apache/kafka/pull/2099.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 #2099


commit 2eff0ac183204a46e67bbef63aa0f4711384f8bc
Author: Ewen Cheslack-Postava 
Date:   2016-11-03T21:54:06Z

MINOR: Fix re-raise of python error in system tests




---
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 #2098: KAFA-4378: resolve eta-expansion of zero-argument ...

[leachbj]

GitHub user leachbj opened a pull request:

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

KAFA-4378: resolve eta-expansion of zero-argument method warnings



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

$ git pull https://github.com/leachbj/kafka 4378-eta-expansion

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

https://github.com/apache/kafka/pull/2098.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 #2098


commit 826f92f520e168d0319917e043af6979b5958a23
Author: Bernard Leach 
Date:   2016-11-01T22:47:21Z

KAFA-4378: resolve eta-expansion of zero-argument method warnings




---
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] [Created] (KAFKA-4378) Address 2.12 eta-expansion warnings

[Bernard Leach (JIRA)]

Bernard Leach created KAFKA-4378:


 Summary: Address 2.12 eta-expansion warnings
 Key: KAFKA-4378
 URL: https://issues.apache.org/jira/browse/KAFKA-4378
 Project: Kafka
  Issue Type: Sub-task
  Components: core
Reporter: Bernard Leach


The 2.12 compiler generates warnings about zero-argument eta-expansion.  Update 
the code to make the expansion explicit.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4377) Address 2.12 deprecations

[Bernard Leach (JIRA)]

Bernard Leach created KAFKA-4377:


 Summary: Address 2.12 deprecations
 Key: KAFKA-4377
 URL: https://issues.apache.org/jira/browse/KAFKA-4377
 Project: Kafka
  Issue Type: Sub-task
  Components: core
Reporter: Bernard Leach


The 2.12 compiler is generating a large number of deprecation warnings, those 
that can be resolved in a 2.10 compatible way should be fixed.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4376) Add scala 2.12 support

[Bernard Leach (JIRA)]

Bernard Leach created KAFKA-4376:


 Summary: Add scala 2.12 support
 Key: KAFKA-4376
 URL: https://issues.apache.org/jira/browse/KAFKA-4376
 Project: Kafka
  Issue Type: Improvement
  Components: build
Affects Versions: 0.10.0.1, 0.10.1.0
Reporter: Bernard Leach


Now that Scala 2.12 has now been officially released releasing 2.12 builds of 
the kafka artifacts will allow downstream projects such as reactive-kafka via 
scalatest-embedded-kafka to release 2.12 builds.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4375) Kafka consumer may swallow some interrupts meant for the calling thread

[JIRA]

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

Stig Rohde Døssing commented on KAFKA-4375:
---

Happy to hear that. I'll try resetting the interrupt flag and see if the tests 
still run.

> Kafka consumer may swallow some interrupts meant for the calling thread
> ---
>
> Key: KAFKA-4375
> URL: https://issues.apache.org/jira/browse/KAFKA-4375
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Stig Rohde Døssing
>
> Apache Storm has added a new data source ("spout") based on the Kafka 0.9 
> consumer. Storm interacts with the consumer by having one thread per spout 
> instance loop calls to poll/commitSync etc. When Storm shuts down, another 
> thread indicates that the looping threads should shut down by interrupting 
> them, and joining them.
> If one of the looping threads happen to be interrupted while executing 
> certain sleeps in some consumer methods (commitSync and committed at least), 
> the interrupt can be lost because they contain a call to SystemTime.sleep, 
> which swallows the interrupt.
> Is this behavior by design, or can SystemTime be changed to reset the thread 
> interrupt flag when catching an InterruptedException? 
> I haven't checked the rest of the client code, so it's possible that this is 
> an issue in other parts of the code too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: [DISCUSS] KIP-82 - Add Record Headers

[Roger Hoover]

 As others have laid out, I see strong reasons for a common message
metadata structure for the Kafka ecosystem.  In particular, I've seen that
even within a single organization, infrastructure teams often own the
message metadata while application teams own the application-level data
format.  Allowing metadata and content to have different structure and
evolve separately is very helpful for this.  Also, I think there's a lot of
value to having a common metadata structure shared across the Kafka
ecosystem so that tools which leverage metadata can more easily be shared
across organizations and integrated together.

The question is, where does the metadata structure belong?  Here's my take:

We change the Kafka wire and on-disk format to from a (key, value) model to
a (key, metadata, value) model where all three are byte arrays from the
brokers point of view.  The primary reason for this is that it provides a
backward compatible migration path forward.  Producers can start populating
metadata fields before all consumers understand the metadata structure.
For people who already have custom envelope structures, they can populate
their existing structure and the new structure for a while as they make the
transition.

We could stop there and let the clients plug in a KeySerializer,
MetadataSerializer, and ValueSerializer but I think it is also be useful to
have a default MetadataSerializer that implements a key-value model similar
to AMQP or HTTP headers.  Or we could go even further and prescribe a
Map or Map data model for headers in the
clients (while still allowing custom serialization of the header data
model).

I think this would address Radai's concerns:
1. All client code would not need to be updated to know about the container.
2. Middleware friendly clients would have a standard header data model to
work with.
3. KIP is required both b/c of broker changes and because of client API
changes.

Cheers,

Roger


On Wed, Nov 2, 2016 at 4:38 PM, radai  wrote:

> my biggest issues with a "standard" wrapper format:
>
> 1. _ALL_ client _CODE_ (as opposed to kafka lib version) must be updated to
> know about the container, because any old naive code trying to directly
> deserialize its own payload would keel over and die (it needs to know to
> deserialize a container, and then dig in there for its payload).
> 2. in order to write middleware-friendly clients that utilize such a
> container one would basically have to write their own producer/consumer API
> on top of the open source kafka one.
> 3. if you were going to go with a wrapper format you really dont need to
> bother with a kip (just open source your own client stack from #2 above so
> others could stop re-inventing it)
>
> On Wed, Nov 2, 2016 at 4:25 PM, James Cheng  wrote:
>
> > How exactly would this work? Or maybe that's out of scope for this email.
>

[jira] [Resolved] (KAFKA-4356) o.a.k.common.utils.SystemTime.sleep() swallows InterruptedException

[Shikhar Bhushan (JIRA)]

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

Shikhar Bhushan resolved KAFKA-4356.

Resolution: Duplicate

> o.a.k.common.utils.SystemTime.sleep() swallows InterruptedException
> ---
>
> Key: KAFKA-4356
> URL: https://issues.apache.org/jira/browse/KAFKA-4356
> Project: Kafka
>  Issue Type: Bug
>Reporter: Shikhar Bhushan
>Priority: Minor
>
> {{org.apache.kafka.common.utils.SystemTime.sleep()}} catches and ignores 
> {{InterruptedException}}. When doing so normally the interruption state 
> should still be restored with {{Thread.currentThread().interrupt()}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4375) Kafka consumer may swallow some interrupts meant for the calling thread

[Shikhar Bhushan (JIRA)]

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

Shikhar Bhushan commented on KAFKA-4375:


Good to have a report of this being a problem, I opened KAFKA-4356 recently 
after chancing on the code. It seems like an oversight rather than by design.

> Kafka consumer may swallow some interrupts meant for the calling thread
> ---
>
> Key: KAFKA-4375
> URL: https://issues.apache.org/jira/browse/KAFKA-4375
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Stig Rohde Døssing
>
> Apache Storm has added a new data source ("spout") based on the Kafka 0.9 
> consumer. Storm interacts with the consumer by having one thread per spout 
> instance loop calls to poll/commitSync etc. When Storm shuts down, another 
> thread indicates that the looping threads should shut down by interrupting 
> them, and joining them.
> If one of the looping threads happen to be interrupted while executing 
> certain sleeps in some consumer methods (commitSync and committed at least), 
> the interrupt can be lost because they contain a call to SystemTime.sleep, 
> which swallows the interrupt.
> Is this behavior by design, or can SystemTime be changed to reset the thread 
> interrupt flag when catching an InterruptedException? 
> I haven't checked the rest of the client code, so it's possible that this is 
> an issue in other parts of the code too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4375) Kafka consumer may swallow some interrupts meant for the calling thread

[JIRA]

Stig Rohde Døssing created KAFKA-4375:
-

 Summary: Kafka consumer may swallow some interrupts meant for the 
calling thread
 Key: KAFKA-4375
 URL: https://issues.apache.org/jira/browse/KAFKA-4375
 Project: Kafka
  Issue Type: Bug
  Components: clients
Reporter: Stig Rohde Døssing


Apache Storm has added a new data source ("spout") based on the Kafka 0.9 
consumer. Storm interacts with the consumer by having one thread per spout 
instance loop calls to poll/commitSync etc. When Storm shuts down, another 
thread indicates that the looping threads should shut down by interrupting 
them, and joining them.

If one of the looping threads happen to be interrupted while executing certain 
sleeps in some consumer methods (commitSync and committed at least), the 
interrupt can be lost because they contain a call to SystemTime.sleep, which 
swallows the interrupt.

Is this behavior by design, or can SystemTime be changed to reset the thread 
interrupt flag when catching an InterruptedException? 

I haven't checked the rest of the client code, so it's possible that this is an 
issue in other parts of the code too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2097: KAFKA-4366: KafkaStreams.close() blocks indefinite...

[dguy]

GitHub user dguy opened a pull request:

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

KAFKA-4366: KafkaStreams.close() blocks indefinitely

Added `timeout` and `timeUnit` to `KafkaStreams.close(..)`. Now do close on 
a thread and `join` that thread with the provided `timeout`.
Changed `state` in `KafkaStreams` to use an enum.
Added system test to ensure we don't deadlock on close when an uncaught 
exception handler that calls `System.exit(..)` is used and there is also a 
shutdown hook that calls `KafkaStreams.close(...)`

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

$ git pull https://github.com/dguy/kafka kafka-4366

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

https://github.com/apache/kafka/pull/2097.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 #2097


commit 6dffa532ca0c065d2e67b812d82b3bdabe97b19d
Author: Damian Guy 
Date:   2016-11-03T18:46:33Z

add timeout to KafkaStreams.close(..)




---
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-4366) KafkaStreams.close() blocks indefinitely

[ASF GitHub Bot (JIRA)]

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

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

GitHub user dguy opened a pull request:

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

KAFKA-4366: KafkaStreams.close() blocks indefinitely

Added `timeout` and `timeUnit` to `KafkaStreams.close(..)`. Now do close on 
a thread and `join` that thread with the provided `timeout`.
Changed `state` in `KafkaStreams` to use an enum.
Added system test to ensure we don't deadlock on close when an uncaught 
exception handler that calls `System.exit(..)` is used and there is also a 
shutdown hook that calls `KafkaStreams.close(...)`

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

$ git pull https://github.com/dguy/kafka kafka-4366

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

https://github.com/apache/kafka/pull/2097.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 #2097


commit 6dffa532ca0c065d2e67b812d82b3bdabe97b19d
Author: Damian Guy 
Date:   2016-11-03T18:46:33Z

add timeout to KafkaStreams.close(..)




> KafkaStreams.close() blocks indefinitely
> 
>
> Key: KAFKA-4366
> URL: https://issues.apache.org/jira/browse/KAFKA-4366
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.1.0, 0.10.0.1
>Reporter: Michal Borowiecki
>Assignee: Damian Guy
>
> KafkaStreams.close() method calls join on all its threads without a timeout, 
> meaning indefinitely, which makes it prone to deadlocks and unfit to be used 
> in shutdown hooks.
> (KafkaStreams::close is used in numerous examples by confluent: 
> https://github.com/confluentinc/examples/tree/kafka-0.10.0.1-cp-3.0.1/kafka-streams/src/main/java/io/confluent/examples/streams
>  and 
> https://www.confluent.io/blog/introducing-kafka-streams-stream-processing-made-simple/
>  so we assumed it to be recommended practice)
> A deadlock happens, for instance, if System.exit() is called from within the 
> uncaughtExceptionHandler. (We need to call System.exit() from the 
> uncaughtExceptionHandler because KAFKA-4355 issue shuts down the StreamThread 
> and to recover we want the process to exit, as our infrastructure will then 
> start it up again.)
> The System.exit call (from the uncaughtExceptionHandler, which runs in the 
> StreamThread) will execute the shutdown hook in a new thread and wait for 
> that thread to join. If the shutdown hook calls KafkaStreams.close, it will 
> in turn block waiting for the StreamThread to join, hence the deadlock.
> Runtime.addShutdownHook javadocs state:
> {quote}
> Shutdown hooks run at a delicate time in the life cycle of a virtual machine 
> and should therefore be coded defensively. They should, in particular, be 
> written to be thread-safe and to avoid deadlocks insofar as possible
> {quote}
> and
> {quote}
> Shutdown hooks should also finish their work quickly.
> {quote}
> Therefore the current implementation of KafkaStreams.close() which waits 
> forever for threads to join is completely unsuitable for use in a shutdown 
> hook. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: [DISCUSS] KAFKA-4345: Run decktape test for each pull request

[Becket Qin]

Thanks for the explanation, Raghav.

If the workload is not a concern then it is probably fine to run tests for
each PR update, although it may not be necessary :)

On Thu, Nov 3, 2016 at 10:40 AM, Raghav Kumar Gautam 
wrote:

> Hi Becket,
>
> The tests would be run each time a PR is created/updated this will look
> similar to https://github.com/apache/storm/pulls. Ducktape tests take
> about
> 7-8 hours to run on my laptop. For travis-ci we can split them in groups
> and run them in parallel. This was done in the POC run which took 1.5 hrs.
> It had 10 splits with 5 jobs running in parallel.
> https://travis-ci.org/raghavgautam/kafka/builds/171502069
> For apache projects the limit is 30 jobs in parallel and across all
> projects, so I expect it to take less time but it also depends on the
> workload at the time.
> https://blogs.apache.org/infra/entry/apache_gains_additional_travis_ci
>
> Thanks,
> Raghav.
>
> On Thu, Nov 3, 2016 at 9:41 AM, Becket Qin  wrote:
>
> > Thanks Raghav,
> >
> > +1 for the idea in general.
> >
> > One thing I am wondering is when the tests would be run? Would it be run
> > when we merge a PR or it would be run every time a PR is created/updated?
> > I am not sure how long do the tests in other projects take. For Kafka it
> > may take a few hours to run all the ducktape tests, will that be an issue
> > if we run the tests for each updates of the PR?
> >
> > Thanks,
> >
> > Jiangjie (Becket) Qin
> >
> > On Thu, Nov 3, 2016 at 8:16 AM, Harsha Chintalapani 
> > wrote:
> >
> > > Thanks, Raghav . I am +1 for having this in Kafka. It will help
> identify
> > > any potential issues, especially with big patches. Given that we've
> some
> > > tests failing due to timing issues
> > > can we disable the failing tests for now so that we don't get any false
> > > negatives?
> > >
> > > Thanks,
> > > Harsha
> > >
> > > On Tue, Nov 1, 2016 at 11:47 AM Raghav Kumar Gautam  >
> > > wrote:
> > >
> > > > Hi,
> > > >
> > > > I want to start a discussion about running ducktape tests for each
> pull
> > > > request. I have been working on KAFKA-4345
> > > >  to enable this
> > using
> > > > docker on travis-ci.
> > > > Pull request: https://github.com/apache/kafka/pull/2064
> > > > Working POC: https://travis-ci.org/raghavgautam/kafka/builds/
> 171502069
> > > >
> > > > In the POC I am able to run 124/149 tests out of which 88 pass. The
> > > failure
> > > > are mostly timing issues. We can run the same scripts on the laptop
> > with
> > > > which I am able to run 138/149 tests successfully.
> > > >
> > > > For this to work we need to enable travis-ci for Kafka. I can open a
> > > infra
> > > > bug to request travis-ci for this. Travis-ci is already running tests
> > for
> > > > many apache projects like Storm, Hive, Flume, Thrift etc. see:
> > > > https://travis-ci.org/apache/.
> > > >
> > > > Does this sound interesting ? Please comment.
> > > >
> > > > Thanks,
> > > > Raghav.
> > > >
> > >
> >
>

[DISCUSS] KIP 88: DescribeGroups Protocol Update

[Vahid S Hashemian]

Hi all,

I started a new KIP under 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-88%3A+DescribeGroups+Protocol+Update
.

The KIP is a proposal to update the DescribeGroups protocol to address 
KAFKA-3853 (https://issues.apache.org/jira/browse/KAFKA-3853).

I appreciate your feedback.

Thanks.
--Vahid

Re: [DISCUSS] KAFKA-4345: Run decktape test for each pull request

[Raghav Kumar Gautam]

Hi Becket,

The tests would be run each time a PR is created/updated this will look
similar to https://github.com/apache/storm/pulls. Ducktape tests take about
7-8 hours to run on my laptop. For travis-ci we can split them in groups
and run them in parallel. This was done in the POC run which took 1.5 hrs.
It had 10 splits with 5 jobs running in parallel.
https://travis-ci.org/raghavgautam/kafka/builds/171502069
For apache projects the limit is 30 jobs in parallel and across all
projects, so I expect it to take less time but it also depends on the
workload at the time.
https://blogs.apache.org/infra/entry/apache_gains_additional_travis_ci

Thanks,
Raghav.

On Thu, Nov 3, 2016 at 9:41 AM, Becket Qin  wrote:

> Thanks Raghav,
>
> +1 for the idea in general.
>
> One thing I am wondering is when the tests would be run? Would it be run
> when we merge a PR or it would be run every time a PR is created/updated?
> I am not sure how long do the tests in other projects take. For Kafka it
> may take a few hours to run all the ducktape tests, will that be an issue
> if we run the tests for each updates of the PR?
>
> Thanks,
>
> Jiangjie (Becket) Qin
>
> On Thu, Nov 3, 2016 at 8:16 AM, Harsha Chintalapani 
> wrote:
>
> > Thanks, Raghav . I am +1 for having this in Kafka. It will help identify
> > any potential issues, especially with big patches. Given that we've some
> > tests failing due to timing issues
> > can we disable the failing tests for now so that we don't get any false
> > negatives?
> >
> > Thanks,
> > Harsha
> >
> > On Tue, Nov 1, 2016 at 11:47 AM Raghav Kumar Gautam 
> > wrote:
> >
> > > Hi,
> > >
> > > I want to start a discussion about running ducktape tests for each pull
> > > request. I have been working on KAFKA-4345
> > >  to enable this
> using
> > > docker on travis-ci.
> > > Pull request: https://github.com/apache/kafka/pull/2064
> > > Working POC: https://travis-ci.org/raghavgautam/kafka/builds/171502069
> > >
> > > In the POC I am able to run 124/149 tests out of which 88 pass. The
> > failure
> > > are mostly timing issues. We can run the same scripts on the laptop
> with
> > > which I am able to run 138/149 tests successfully.
> > >
> > > For this to work we need to enable travis-ci for Kafka. I can open a
> > infra
> > > bug to request travis-ci for this. Travis-ci is already running tests
> for
> > > many apache projects like Storm, Hive, Flume, Thrift etc. see:
> > > https://travis-ci.org/apache/.
> > >
> > > Does this sound interesting ? Please comment.
> > >
> > > Thanks,
> > > Raghav.
> > >
> >
>

Re: [DISCUSS] KAFKA-4345: Run decktape test for each pull request

[Becket Qin]

Thanks Raghav,

+1 for the idea in general.

One thing I am wondering is when the tests would be run? Would it be run
when we merge a PR or it would be run every time a PR is created/updated?
I am not sure how long do the tests in other projects take. For Kafka it
may take a few hours to run all the ducktape tests, will that be an issue
if we run the tests for each updates of the PR?

Thanks,

Jiangjie (Becket) Qin

On Thu, Nov 3, 2016 at 8:16 AM, Harsha Chintalapani  wrote:

> Thanks, Raghav . I am +1 for having this in Kafka. It will help identify
> any potential issues, especially with big patches. Given that we've some
> tests failing due to timing issues
> can we disable the failing tests for now so that we don't get any false
> negatives?
>
> Thanks,
> Harsha
>
> On Tue, Nov 1, 2016 at 11:47 AM Raghav Kumar Gautam 
> wrote:
>
> > Hi,
> >
> > I want to start a discussion about running ducktape tests for each pull
> > request. I have been working on KAFKA-4345
> >  to enable this using
> > docker on travis-ci.
> > Pull request: https://github.com/apache/kafka/pull/2064
> > Working POC: https://travis-ci.org/raghavgautam/kafka/builds/171502069
> >
> > In the POC I am able to run 124/149 tests out of which 88 pass. The
> failure
> > are mostly timing issues. We can run the same scripts on the laptop with
> > which I am able to run 138/149 tests successfully.
> >
> > For this to work we need to enable travis-ci for Kafka. I can open a
> infra
> > bug to request travis-ci for this. Travis-ci is already running tests for
> > many apache projects like Storm, Hive, Flume, Thrift etc. see:
> > https://travis-ci.org/apache/.
> >
> > Does this sound interesting ? Please comment.
> >
> > Thanks,
> > Raghav.
> >
>

[jira] [Created] (KAFKA-4374) Improve Response Errors Logging

[Jesse Anderson (JIRA)]

Jesse Anderson created KAFKA-4374:
-

 Summary: Improve Response Errors Logging
 Key: KAFKA-4374
 URL: https://issues.apache.org/jira/browse/KAFKA-4374
 Project: Kafka
  Issue Type: Bug
  Components: clients
Affects Versions: 0.10.0.1
Reporter: Jesse Anderson
Priority: Minor


When NetworkClient.java gets a response error, it runs:
{code}
if (response.errors().size() > 0) {
log.warn("Error while fetching metadata with correlation id {} 
: {}", header.correlationId(), response.errors());
}
{code}

Logging that at warn level and saying there is an error, is confusing to new 
people. They don't see it was a warn and not error level. They just see that it 
says "Error while...".

Maybe it should be something like "The metadata response from the cluster 
reported a recoverable issue..."



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: [ANNOUNCE] New committer: Jiangjie (Becket) Qin

[Efe Gencer]

Congratulations, Becket!

Best,
Efe

2016-11-03 11:22 GMT-04:00 Neha Narkhede :

> Congratulations, Becket! Well done.
>
> On Wed, Nov 2, 2016 at 1:34 PM Eno Thereska 
> wrote:
>
> > Congrats!
> > Eno
> >
> > > On 1 Nov 2016, at 05:57, Harsha Chintalapani  wrote:
> > >
> > > Congrats Becket!
> > > -Harsha
> > >
> > > On Mon, Oct 31, 2016 at 2:13 PM Rajini Sivaram <
> > rajinisiva...@googlemail.com>
> > > wrote:
> > >
> > >> Congratulations, Becket!
> > >>
> > >> On Mon, Oct 31, 2016 at 8:38 PM, Matthias J. Sax <
> matth...@confluent.io
> > >
> > >> wrote:
> > >>
> > >>> -BEGIN PGP SIGNED MESSAGE-
> > >>> Hash: SHA512
> > >>>
> > >>> Congrats!
> > >>>
> > >>> On 10/31/16 11:01 AM, Renu Tewari wrote:
> >  Congratulations Becket!! Absolutely thrilled to hear this. Well
> >  deserved!
> > 
> >  regards renu
> > 
> > 
> >  On Mon, Oct 31, 2016 at 10:35 AM, Joel Koshy 
> >  wrote:
> > 
> > > The PMC for Apache Kafka has invited Jiangjie (Becket) Qin to
> > > join as a committer and we are pleased to announce that he has
> > > accepted!
> > >
> > > Becket has made significant contributions to Kafka over the last
> > > two years. He has been deeply involved in a broad range of KIP
> > > discussions and has contributed several major features to the
> > > project. He recently completed the implementation of a series of
> > > improvements (KIP-31, KIP-32, KIP-33) to Kafka’s message format
> > > that address a number of long-standing issues such as avoiding
> > > server-side re-compression, better accuracy for time-based log
> > > retention, log roll and time-based indexing of messages.
> > >
> > > Congratulations Becket! Thank you for your many contributions. We
> > > are excited to have you on board as a committer and look forward
> > > to your continued participation!
> > >
> > > Joel
> > >
> > 
> > >>> -BEGIN PGP SIGNATURE-
> > >>> Comment: GPGTools - https://gpgtools.org
> > >>>
> > >>> iQIcBAEBCgAGBQJYF6uzAAoJECnhiMLycopPBuwP/1N2MtwWw7ms5gAfT/jvVCGi
> > >>> mdNvdJprSwJHe3qwsc+glsvAqwS6OZfaVzK2qQcaxMX5KjQtwkkOKyErOl9hG7jD
> > >>> Vw0aDcCbPuV2oEZ4m9K2J4Q3mZIfFrevicVb7oPGf4Yjt1sh9wxP08o7KHP2l5pN
> > >>> 3mpIBEDp4rZ2pg/jXldyh57dW1btg3gZi1gNczWvXEAKf1ypXRPwPeDbvXADXDv3
> > >>> 0NgmcXn242geoggnIbL30WgjH0bwHpVjLBr++YQ33FzRoHzASfAYHR/jSDKAytQe
> > >>> a7Bkc69Bb1NSzkfhiJa+VW9V2DweO8kD+Xfz4dM02GQF0iJkAqare7a6zWedk/+U
> > >>> hJRPz+tGlDSLePCYdyNj1ivJrFOmIQtyFOI3SBANfaneOmGJhPKtlNQQlNFKDbWS
> > >>> CD1pBsc1iHNq6rXy21evc/aFk0Rrfs5d4rU9eG6jD8jc1mCbSwtzJI0vweX0r9Y/
> > >>> 6Ao8cnsmDejYfap5lUMWeQfZOTkNRNpbkL7eoiVpe6wZw1nGL3T7GkrrWGRS3EQO
> > >>> qp4Jjp+7yY4gIqsLfYouaHTEzAX7yN78QNUNCB4OqUiEL9+a8wTQ7dlTgXinEd8r
> > >>> Kh9vTfpW7fb4c58aSpzntPUU4YFD3MHMam0iu5UrV9d5DrVTFDMJ83k15Z5DyTMt
> > >>> 45nPYdjvJgFGWLYFnPwr
> > >>> =VbpG
> > >>> -END PGP SIGNATURE-
> > >>>
> > >>
> > >>
> > >>
> > >> --
> > >> Regards,
> > >>
> > >> Rajini
> > >>
> >
> > --
> Thanks,
> Neha
>

Re: [DISCUSS] KIP-84: Support SASL/SCRAM mechanisms

[Harsha Chintalapani]

Agree with Rajini on not incrementing the protocol version. As brokers are
returning the list of supported mechanisms don't think it warrants a
protocol version bump.

Thanks,
Harsha

On Thu, Nov 3, 2016 at 7:59 AM Rajini Sivaram 
wrote:

> Ismael,
>
> 2) I left out MD5 because it is insecure, but thought of keeping SHA-1
> since it is supported by most services that support SCRAM today. Since
> there is no actual requirement to support SHA-1 and we don't really want to
> store credentials in Zookeeper using SHA-1, it makes sense not to support
> SHA-1. Will update the KIP. Thanks.
>
> On Thu, Nov 3, 2016 at 1:25 PM, Ismael Juma  wrote:
>
> > Hi all,
> >
> > I'm currently with limited internet access so I have not done a proper
> > review of the KIP, my apologies. A couple of thoughts:
> >
> > 1. I agree with Rajini and I don't see the value in bumping the protocol
> > version in this case. As Rajini said, the enabled mechanisms are
> > independent of the broker version. If and when we have a feature API, we
> > can consider exposing it that way.
> >
> > 2. Do we really want to support sha1? The security of it has been
> > questioned and the general recommendation is to avoid it these days.
> >
> > Ismael
> >
> > On 3 Nov 2016 6:51 am, "Rajini Sivaram" 
> > wrote:
> >
> > > I don't have a strong objection to bumping up SaslHandshakeRequest
> > version,
> > > though I am still not convinced of the usefulness. I do agree that
> KIP-35
> > > should be standard way to determine client/broker compatibility. But I
> am
> > > not sure ApiVersionsRequest is a good fit for clients to choose a SASL
> > > mechanism.
> > >
> > >1. SaslHandshakeRequest is the only way a client can figure out if a
> > >SASL mechanism is actually enabled in the broker. The fact that
> broker
> > >version n supports SCRAM doesn't imply that a broker of version n
> > > actually
> > >has the mechanism enabled. Since enabling a mechanism involves some
> > > effort
> > >like installing an authentication server and/or configuring
> > credentials,
> > >SASL mechanisms are a feature of a broker installation very unlike
> > >versions.  As you said, "features" would have worked better.
> > >2. New SASL mechanisms can be added to older Kafka broker versions.
> > With
> > >some tweaking, the SCRAM implementation from KIP-84 can be enabled
> in
> > a
> > >0.10.0 broker without changing any broker code. KIP-86 would make
> this
> > > even
> > >easier, but it is already possible to add new or custom mechanisms
> to
> > >existing broker versions. A client using ApiVersionsRequest to
> choose
> > a
> > >SASL mechanism is only checking when a mechanism was included in a
> > > broker
> > >release, not the "capability" of a broker to support the mechanism.
> I
> > am
> > >not sure we should encourage clients to choose mechanisms based on
> > > versions.
> > >3. Clients need additional configuration based on the chosen
> > mechanism.
> > >One of the reasons I couldn't see any value in using
> > ApiVersionsRequest
> > > in
> > >the Java client was because clients are configured with a single
> SASL
> > >mechanism and a JAAS configuration corresponding to that mechanism.
> > If a
> > >client wants to choose between Kerberos and SCRAM, the client would
> > need
> > >keytab/principal for kerberos and username/password for SCRAM.
> Clients
> > > that
> > >possess multiple credentials without knowing what the broker
> supports
> > -
> > >that sounds like a rather unusual scenario.
> > >4. For some mechanisms, clients may want to choose mechanism at
> > runtime.
> > >For example, a client is configured with username/ password and
> wants
> > to
> > >choose between SCRAM-SHA-1/SCRAM-SHA-256/PLAIN depending on which
> > >mechanisms are enabled in the broker. To choose between SCRAM-SHA-1
> > and
> > >SCRAM-SHA-256, clients have to use SaslHandshakeRequest since they
> > were
> > >added in the same version. To choose between PLAIN and
> SCRAM-SHA-256,
> > a
> > >versions based approach would work, but wouldn't it be better for
> > > clients
> > >to rely on just SaslHandshakeRequest rather than
> > >ApiVersionsRequest+SaslHandshakeRequest so that the solution works
> in
> > > all
> > >scenarios?
> > >
> > >
> > >
> > >
> > > On Thu, Nov 3, 2016 at 4:33 AM, Ewen Cheslack-Postava <
> e...@confluent.io
> > >
> > > wrote:
> > >
> > > > I think the bump isn't strictly required, but if the client is KIP-35
> > > > aware, it can proactively choose a compatible SASL mechanism based on
> > its
> > > > initial ApiVersionRequest and avoid an extra connection round trip
> when
> > > > there are client/broker version differences. Without this, a newer
> > client
> > > > would have to do 2 set of requests since the first SASL mechanism
> might
> > > not
> > > > be 

Re: [ANNOUNCE] New committer: Jiangjie (Becket) Qin

[Neha Narkhede]

Congratulations, Becket! Well done.

On Wed, Nov 2, 2016 at 1:34 PM Eno Thereska  wrote:

> Congrats!
> Eno
>
> > On 1 Nov 2016, at 05:57, Harsha Chintalapani  wrote:
> >
> > Congrats Becket!
> > -Harsha
> >
> > On Mon, Oct 31, 2016 at 2:13 PM Rajini Sivaram <
> rajinisiva...@googlemail.com>
> > wrote:
> >
> >> Congratulations, Becket!
> >>
> >> On Mon, Oct 31, 2016 at 8:38 PM, Matthias J. Sax  >
> >> wrote:
> >>
> >>> -BEGIN PGP SIGNED MESSAGE-
> >>> Hash: SHA512
> >>>
> >>> Congrats!
> >>>
> >>> On 10/31/16 11:01 AM, Renu Tewari wrote:
>  Congratulations Becket!! Absolutely thrilled to hear this. Well
>  deserved!
> 
>  regards renu
> 
> 
>  On Mon, Oct 31, 2016 at 10:35 AM, Joel Koshy 
>  wrote:
> 
> > The PMC for Apache Kafka has invited Jiangjie (Becket) Qin to
> > join as a committer and we are pleased to announce that he has
> > accepted!
> >
> > Becket has made significant contributions to Kafka over the last
> > two years. He has been deeply involved in a broad range of KIP
> > discussions and has contributed several major features to the
> > project. He recently completed the implementation of a series of
> > improvements (KIP-31, KIP-32, KIP-33) to Kafka’s message format
> > that address a number of long-standing issues such as avoiding
> > server-side re-compression, better accuracy for time-based log
> > retention, log roll and time-based indexing of messages.
> >
> > Congratulations Becket! Thank you for your many contributions. We
> > are excited to have you on board as a committer and look forward
> > to your continued participation!
> >
> > Joel
> >
> 
> >>> -BEGIN PGP SIGNATURE-
> >>> Comment: GPGTools - https://gpgtools.org
> >>>
> >>> iQIcBAEBCgAGBQJYF6uzAAoJECnhiMLycopPBuwP/1N2MtwWw7ms5gAfT/jvVCGi
> >>> mdNvdJprSwJHe3qwsc+glsvAqwS6OZfaVzK2qQcaxMX5KjQtwkkOKyErOl9hG7jD
> >>> Vw0aDcCbPuV2oEZ4m9K2J4Q3mZIfFrevicVb7oPGf4Yjt1sh9wxP08o7KHP2l5pN
> >>> 3mpIBEDp4rZ2pg/jXldyh57dW1btg3gZi1gNczWvXEAKf1ypXRPwPeDbvXADXDv3
> >>> 0NgmcXn242geoggnIbL30WgjH0bwHpVjLBr++YQ33FzRoHzASfAYHR/jSDKAytQe
> >>> a7Bkc69Bb1NSzkfhiJa+VW9V2DweO8kD+Xfz4dM02GQF0iJkAqare7a6zWedk/+U
> >>> hJRPz+tGlDSLePCYdyNj1ivJrFOmIQtyFOI3SBANfaneOmGJhPKtlNQQlNFKDbWS
> >>> CD1pBsc1iHNq6rXy21evc/aFk0Rrfs5d4rU9eG6jD8jc1mCbSwtzJI0vweX0r9Y/
> >>> 6Ao8cnsmDejYfap5lUMWeQfZOTkNRNpbkL7eoiVpe6wZw1nGL3T7GkrrWGRS3EQO
> >>> qp4Jjp+7yY4gIqsLfYouaHTEzAX7yN78QNUNCB4OqUiEL9+a8wTQ7dlTgXinEd8r
> >>> Kh9vTfpW7fb4c58aSpzntPUU4YFD3MHMam0iu5UrV9d5DrVTFDMJ83k15Z5DyTMt
> >>> 45nPYdjvJgFGWLYFnPwr
> >>> =VbpG
> >>> -END PGP SIGNATURE-
> >>>
> >>
> >>
> >>
> >> --
> >> Regards,
> >>
> >> Rajini
> >>
>
> --
Thanks,
Neha

Re: [DISCUSS] KAFKA-4345: Run decktape test for each pull request

[Harsha Chintalapani]

Thanks, Raghav . I am +1 for having this in Kafka. It will help identify
any potential issues, especially with big patches. Given that we've some
tests failing due to timing issues
can we disable the failing tests for now so that we don't get any false
negatives?

Thanks,
Harsha

On Tue, Nov 1, 2016 at 11:47 AM Raghav Kumar Gautam 
wrote:

> Hi,
>
> I want to start a discussion about running ducktape tests for each pull
> request. I have been working on KAFKA-4345
>  to enable this using
> docker on travis-ci.
> Pull request: https://github.com/apache/kafka/pull/2064
> Working POC: https://travis-ci.org/raghavgautam/kafka/builds/171502069
>
> In the POC I am able to run 124/149 tests out of which 88 pass. The failure
> are mostly timing issues. We can run the same scripts on the laptop with
> which I am able to run 138/149 tests successfully.
>
> For this to work we need to enable travis-ci for Kafka. I can open a infra
> bug to request travis-ci for this. Travis-ci is already running tests for
> many apache projects like Storm, Hive, Flume, Thrift etc. see:
> https://travis-ci.org/apache/.
>
> Does this sound interesting ? Please comment.
>
> Thanks,
> Raghav.
>

Re: [DISCUSS] KIP-84: Support SASL/SCRAM mechanisms

[Rajini Sivaram]

Ismael,

2) I left out MD5 because it is insecure, but thought of keeping SHA-1
since it is supported by most services that support SCRAM today. Since
there is no actual requirement to support SHA-1 and we don't really want to
store credentials in Zookeeper using SHA-1, it makes sense not to support
SHA-1. Will update the KIP. Thanks.

On Thu, Nov 3, 2016 at 1:25 PM, Ismael Juma  wrote:

> Hi all,
>
> I'm currently with limited internet access so I have not done a proper
> review of the KIP, my apologies. A couple of thoughts:
>
> 1. I agree with Rajini and I don't see the value in bumping the protocol
> version in this case. As Rajini said, the enabled mechanisms are
> independent of the broker version. If and when we have a feature API, we
> can consider exposing it that way.
>
> 2. Do we really want to support sha1? The security of it has been
> questioned and the general recommendation is to avoid it these days.
>
> Ismael
>
> On 3 Nov 2016 6:51 am, "Rajini Sivaram" 
> wrote:
>
> > I don't have a strong objection to bumping up SaslHandshakeRequest
> version,
> > though I am still not convinced of the usefulness. I do agree that KIP-35
> > should be standard way to determine client/broker compatibility. But I am
> > not sure ApiVersionsRequest is a good fit for clients to choose a SASL
> > mechanism.
> >
> >1. SaslHandshakeRequest is the only way a client can figure out if a
> >SASL mechanism is actually enabled in the broker. The fact that broker
> >version n supports SCRAM doesn't imply that a broker of version n
> > actually
> >has the mechanism enabled. Since enabling a mechanism involves some
> > effort
> >like installing an authentication server and/or configuring
> credentials,
> >SASL mechanisms are a feature of a broker installation very unlike
> >versions.  As you said, "features" would have worked better.
> >2. New SASL mechanisms can be added to older Kafka broker versions.
> With
> >some tweaking, the SCRAM implementation from KIP-84 can be enabled in
> a
> >0.10.0 broker without changing any broker code. KIP-86 would make this
> > even
> >easier, but it is already possible to add new or custom mechanisms to
> >existing broker versions. A client using ApiVersionsRequest to choose
> a
> >SASL mechanism is only checking when a mechanism was included in a
> > broker
> >release, not the "capability" of a broker to support the mechanism. I
> am
> >not sure we should encourage clients to choose mechanisms based on
> > versions.
> >3. Clients need additional configuration based on the chosen
> mechanism.
> >One of the reasons I couldn't see any value in using
> ApiVersionsRequest
> > in
> >the Java client was because clients are configured with a single SASL
> >mechanism and a JAAS configuration corresponding to that mechanism.
> If a
> >client wants to choose between Kerberos and SCRAM, the client would
> need
> >keytab/principal for kerberos and username/password for SCRAM. Clients
> > that
> >possess multiple credentials without knowing what the broker supports
> -
> >that sounds like a rather unusual scenario.
> >4. For some mechanisms, clients may want to choose mechanism at
> runtime.
> >For example, a client is configured with username/ password and wants
> to
> >choose between SCRAM-SHA-1/SCRAM-SHA-256/PLAIN depending on which
> >mechanisms are enabled in the broker. To choose between SCRAM-SHA-1
> and
> >SCRAM-SHA-256, clients have to use SaslHandshakeRequest since they
> were
> >added in the same version. To choose between PLAIN and SCRAM-SHA-256,
> a
> >versions based approach would work, but wouldn't it be better for
> > clients
> >to rely on just SaslHandshakeRequest rather than
> >ApiVersionsRequest+SaslHandshakeRequest so that the solution works in
> > all
> >scenarios?
> >
> >
> >
> >
> > On Thu, Nov 3, 2016 at 4:33 AM, Ewen Cheslack-Postava  >
> > wrote:
> >
> > > I think the bump isn't strictly required, but if the client is KIP-35
> > > aware, it can proactively choose a compatible SASL mechanism based on
> its
> > > initial ApiVersionRequest and avoid an extra connection round trip when
> > > there are client/broker version differences. Without this, a newer
> client
> > > would have to do 2 set of requests since the first SASL mechanism might
> > not
> > > be compatible.
> > >
> > > I don't think this is a deal breaker, but I do think it would be good
> to
> > > just standardize on KIP-35 as the way we figure out client/broker
> > > compatibility. The SASL stuff happened in parallel (maybe before?)
> KIP-35
> > > and ended up with its own mechanism, but I'm in favor of trying to
> > simplify
> > > everything by centralizing those considerations into a single API call.
> > (By
> > > the way, dredging up now ancient history in the KIP-35 discussion, this
> > is
> > > also why 

Jenkins build is back to normal : kafka-trunk-jdk7 #1669

[Apache Jenkins Server]

See 

[jira] [Created] (KAFKA-4373) Kafka Consumer API jumping offsets

[Srinivasan Venkatraman (JIRA)]

Srinivasan Venkatraman created KAFKA-4373:
-

 Summary: Kafka Consumer API jumping offsets
 Key: KAFKA-4373
 URL: https://issues.apache.org/jira/browse/KAFKA-4373
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Reporter: Srinivasan Venkatraman


Hi,

I am using Kafka Version 0.10.0.1 and java consumer API to consume messages 
from a topic. We are using a single node kafka and zookeeper. It is sometimes 
observed that the consumer is losing a bulk of message. We are unable to find 
the exact reason to replicate the issue.
The scenario is:
Consumer polls the topic.
Fetches the messages and gives it to a thread pool to handle the message.
Waits for the threads to return and then commits the offsets.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Re: [DISCUSS] KIP-87 - Add Compaction Tombstone Flag

[Becket Qin]

Hi Michael,

Do you mean using a new configuration it is just the exiting
message.format.version config? It seems the message.format.version config
is enough in this case. And the default value would always be the latest
version.

> Message version migration would be handled as like in KIP-32

Also just want to confirm on this. Today if an old consumer consumes a log
compacted topic and sees an empty value, it knows that is a tombstone.
After we start to use the attribute bit, a tombstone message can have a
non-empty value. So by "like in KIP-32" you mean we will remove the value
to down convert the message if the consumer version is old, right?

Thanks.

Jiangjie (Becket) Qin

On Wed, Nov 2, 2016 at 1:37 AM, Michael Pearce 
wrote:

> Hi Joel , et al.
>
> Any comments on the below idea to handle roll out / compatibility of this
> feature, using a configuration?
>
> Does it make sense/clear?
> Does it add value?
> Do we want to enforce flag by default, or value by default, or both?
>
> Cheers
> Mike
>
>
> On 10/27/16, 4:47 PM, "Michael Pearce"  wrote:
>
> Thanks, James, I think this is a really good addition to the KIP
> details, please feel free to amend the wiki/add the use cases, also if any
> others you think of. I definitely think its worthwhile documenting. If you
> can’t let me know ill add them next week (just leaving for a long weekend
> off)
>
> Re Joel and others comments about upgrade and compatibility.
>
> Rather than trying to auto manage this.
>
> Actually maybe we make a configuration option, both at server and per
> topic level to control the behavior of how the server logic should work out
> if the record, is a tombstone record .
>
> e.g.
>
> key = compation.tombstone.marker
>
> value options:
>
> value   (continues to use null value as tombstone marker)
> flag (expects to use the tombstone flag)
> value_or_flag (if either is true it treats the record as a tombstone)
>
> This way on upgrade users can keep current behavior, and slowly
> migrate to the new. Having a transition period of using value_or_flag,
> finally having flag only if an organization wishes to use null values
> without it being treated as a tombstone marker (use case noted below)
>
> Having it both global broker level and topic override also allows some
> flexibility here.
>
> Cheers
> Mike
>
>
>
>
>
>
> On 10/27/16, 8:03 AM, "James Cheng"  wrote:
>
> This KIP would definitely address a gap in the current
> functionality, where you currently can't have a tombstone with any
> associated content.
>
> That said, I'd like to talk about use cases, to make sure that
> this is in fact useful. The KIP should be updated with whatever use cases
> we come up with.
>
> First of all, an observation: When we speak about log compaction,
> we typically think of "the latest message for a key is retained". In that
> respect, a delete tombstone (i.e. a message with a null payload) is treated
> the same as any other Kafka message: the latest message is retained. It
> doesn't matter whether the latest message is null, or if the latest message
> has actual content. In all cases, the last message is retained.
>
> The only way a delete tombstone is treated differently from other
> Kafka messages is that it automatically disappears after a while. The time
> of deletion is specified using delete.retention.ms.
>
> So what we're really talking about is, do we want to support
> messages in a log-compacted topic that auto-delete themselves after a while?
>
> In a thread from 2015, there was a discussion on first-class
> support of headers between Roger Hoover, Felix GV, Jun Rao, and I. See
> thread at https://groups.google.com/d/msg/confluent-platform/
> 8xPbjyUE_7E/yQ1AeCufL_gJ  msg/confluent-platform/8xPbjyUE_7E/yQ1AeCufL_gJ> . In that thread, Jun
> raised a good question that I didn't have a good answer for at the time: If
> a message is going to auto-delete itself after a while, how important was
> the message? That is, what information did the message contain that was
> important *for a while* but not so important that it needed to be kept
> around forever?
>
> Some use cases that I can think of:
>
> 1) Tracability. I would like to know who issued this delete
> tombstone. It might include the hostname, IP of the producer of the delete.
> 2) Timestamps. I would like to know when this delete was issued.
> This use case is already addressed by the availability of per-message
> timestamps that came in 0.10.0
> 3) Data provenance. I hope I'm using this phrase correctly, but
> what I mean is, where did this delete come from? What processing job
> emitted it? What input to the processing job caused this delete to be
> produced? For example, if a record in topic A was processed and caused a
> delete tombstone to be 

RE: Kafka Connect key.converter and value.converter properties for Avro encoding

[david.franklin]

Thanks to Gwen and Tommy Baker for their helpful replies.

Currently, the environment I need to work with doesn't use the Schema Registry; 
hopefully one day it will but for now that's not an option.  Events are written 
to Kafka without the schema embedded and each side of the interface assumes a 
given schema, with the consequent risks accepted.

To serialize a SpecificRecord for the 
org.apache.kafka.connect.storage.Converter interface (in the absence of access 
to the Confluent implementation classes) I was thinking of something along 
these lines to Avro encode a SpecificRecord:

private byte[] toAvro(Schema schema, SpecificRecord record) throws 
IOException{
SpecificDatumWriter writer = new 
SpecificDatumWriter<>(schema);
ByteArrayOutputStream baos = new ByteArrayOutputStream();
BinaryEncoder binaryEncoder = null;
binaryEncoder = new EncoderFactory().binaryEncoder(baos, binaryEncoder);
writer.write(record, binaryEncoder);
return baos.toByteArray();
}

To work with Kafka Connect I need to comply with the 
org.apache.kafka.connect.storage .Converter interface
The Converter interface defines the following methods:

void configure(Map configs, boolean isKey);
byte[] fromConnectData(String topic, Schema schema, Object value);
SchemaAndValue toConnectData(String topic, byte[] value);

Is it safe to provide a no-op implementation for configure().

The toConnectData() method will presumably be achieved via a corresponding 
SpecificDatumReader.

Does this look a reasonable approach?

Many thanks if you've read this far!

Regards,
David


-Original Message-
From: Gwen Shapira [mailto:g...@confluent.io] 
Sent: 02 November 2016 21:18
To: dev@kafka.apache.org
Subject: Re: Kafka Connect key.converter and value.converter properties for 
Avro encoding

Both the Confluent Avro Converter and the Confluent Avro Serializer use the 
Schema Registry. The reason is, as Tommy Becker mentioned below, to avoid 
storing the entire schema in each record (which the JSON serializer in Apache 
Kafka does). It has few other benefits schema validation and such.

If you are interested in trying this approach, you will want to use the 
Converter, since it was written specifically to integrate with Connect.
If you prefer another approach, without the Schema Registry, you can write your 
own Converter - that's why we made them pluggable. Feel free to copy ours and 
modify it as fits your Avro approach.

Gwen

On Wed, Nov 2, 2016 at 2:48 AM,  wrote:

> I am using Kafka Connect in source mode i.e. using it to send events 
> to Kafka topics.
>
> With the key.converter and value.converter properties set to 
> org.apache.kafka.connect.storage.StringConverter I can attach a 
> consumer to the topics and see the events in a readable form.  This is 
> helpful and reassuring but it is not the desired representation for my 
> downstream consumers - these require the events to be Avro encoded.
>
> It seems that to write the events to Kafka Avro encoded, these 
> properties need to be set to 
> io.confluent.kafka.serializers.KafkaAvroSerializer.  Is this correct?
>
> I am not using the Confluent platform, merely the standard Kafka 10 
> download, and have been unable to find out how to get at these from a 
> Maven repository jar.  
> http://docs.confluent.io/3.0.0/app-development.html#java
> suggest that these are available via:
>
>
>  io.confluent
>  kafka-avro-serializer
>  3.0.0
>  
>
> But it doesn't appear to be true.  The class exists in
> https://raw.githubusercontent.com/confluentinc/schema-
> registry/master/avro-converter/src/main/java/io/confluent/connect/avro
> / AvroConverter.java but this seems to use the Schema Registry which 
> is something I'd rather avoid.
>
> I'd be grateful for any pointers on the simplest way of getting Avro 
> encoded events written to Kafka from a Kafka Connect source connector/task.
>
> Also in the task which creates SourceRecords, I'm choosing 
> Schema.BYTES_SCHEMA for the 4th arg in the constructor.  But I'm not 
> clear what this achieves - some light shed on that would also be helpful.
>
> Many thanks,
> David
>



--
*Gwen Shapira*
Product Manager | Confluent
650.450.2760 | @gwenshap
Follow us: Twitter  | blog 

[jira] [Commented] (KAFKA-4372) Kafka Connect REST API does not handle DELETE of connector with slashes in their names

[ASF GitHub Bot (JIRA)]

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

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

GitHub user ogirardot opened a pull request:

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

KAFKA-4372: Kafka Connect REST API does not handle DELETE of connector with 
slashes in their names

Kafka Connect REST API does not handle in many places connectors with 
slashes in their names because it expects PathParams, this PR intends to : 
* Reject as bad requests API calls trying to create connectors with slashes 
in their names
* Add support for connector with slashes in their names in the DELETE part 
of the API to allow users to cleanup their connectors without dropping 
everything.

This PR adds as well the Unit Test needed for the creation part and was 
tested manually for the DELETE part.

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

$ git pull https://github.com/ogirardot/kafka 
fix/connectors-with-slashes-cannot-be-deleted

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

https://github.com/apache/kafka/pull/2096.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 #2096


commit cfdc2a9e50ae75bcaaa98cebf77b91c3fde963f3
Author: Olivier Girardot 
Date:   2016-11-03T13:24:12Z

Fix the delete kafka connectors REST API that did not accept slashes

Even though the rest of the API points accept connector names containing 
slashes

commit eb08fbbca95ca9bdafc4058ada563b7d1615a721
Author: Olivier Girardot 
Date:   2016-11-03T13:52:30Z

Connectors should no longer have slashes in their names

Added a BadRequestException if someone tries to create one like that
and added unit test to check that




> Kafka Connect REST API does not handle DELETE of connector with slashes in 
> their names
> --
>
> Key: KAFKA-4372
> URL: https://issues.apache.org/jira/browse/KAFKA-4372
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.0.0, 0.10.0.1
>Reporter: Olivier Girardot
>Assignee: Ewen Cheslack-Postava
>
> Currently there is nothing to prevent someone from registering a Kafka 
> Connector with slashes in its name, however it's impossible to DELETE it 
> afterwards because the DELETE REST API access point is using a PathParam and 
> does not allow slashes.
> A few other API points will have a tough times handling connectors with 
> slashes in their names.
> We should allow for slashes in the DELETE API points to allow current setups 
> to be cleaned up without having to drop all the other connectors, and not 
> allow anymore connectors to be created with slashes in their names.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2096: KAFKA-4372: Kafka Connect REST API does not handle...

[ogirardot]

GitHub user ogirardot opened a pull request:

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

KAFKA-4372: Kafka Connect REST API does not handle DELETE of connector with 
slashes in their names

Kafka Connect REST API does not handle in many places connectors with 
slashes in their names because it expects PathParams, this PR intends to : 
* Reject as bad requests API calls trying to create connectors with slashes 
in their names
* Add support for connector with slashes in their names in the DELETE part 
of the API to allow users to cleanup their connectors without dropping 
everything.

This PR adds as well the Unit Test needed for the creation part and was 
tested manually for the DELETE part.

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

$ git pull https://github.com/ogirardot/kafka 
fix/connectors-with-slashes-cannot-be-deleted

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

https://github.com/apache/kafka/pull/2096.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 #2096


commit cfdc2a9e50ae75bcaaa98cebf77b91c3fde963f3
Author: Olivier Girardot 
Date:   2016-11-03T13:24:12Z

Fix the delete kafka connectors REST API that did not accept slashes

Even though the rest of the API points accept connector names containing 
slashes

commit eb08fbbca95ca9bdafc4058ada563b7d1615a721
Author: Olivier Girardot 
Date:   2016-11-03T13:52:30Z

Connectors should no longer have slashes in their names

Added a BadRequestException if someone tries to create one like that
and added unit test to check that




---
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] [Created] (KAFKA-4372) Kafka Connect REST API does not handle DELETE of connector with slashes in their names

[Olivier Girardot (JIRA)]

Olivier Girardot created KAFKA-4372:
---

 Summary: Kafka Connect REST API does not handle DELETE of 
connector with slashes in their names
 Key: KAFKA-4372
 URL: https://issues.apache.org/jira/browse/KAFKA-4372
 Project: Kafka
  Issue Type: Bug
  Components: KafkaConnect
Affects Versions: 0.10.0.1, 0.10.0.0
Reporter: Olivier Girardot
Assignee: Ewen Cheslack-Postava


Currently there is nothing to prevent someone from registering a Kafka 
Connector with slashes in its name, however it's impossible to DELETE it 
afterwards because the DELETE REST API access point is using a PathParam and 
does not allow slashes.

A few other API points will have a tough times handling connectors with slashes 
in their names.

We should allow for slashes in the DELETE API points to allow current setups to 
be cleaned up without having to drop all the other connectors, and not allow 
anymore connectors to be created with slashes in their names.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4371) Sporadic ConnectException shuts down the whole connect process

[Sagar Rao (JIRA)]

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

Sagar Rao commented on KAFKA-4371:
--

Another thing to note is that we get this exception when I try to delete an 
existing connector. After that we get a CommunicationLinkException and then all 
the processes stop. Not sure how is this connected though.

> Sporadic ConnectException shuts down the whole connect process
> --
>
> Key: KAFKA-4371
> URL: https://issues.apache.org/jira/browse/KAFKA-4371
> Project: Kafka
>  Issue Type: Bug
>Reporter: Sagar Rao
>Priority: Critical
>
> I had setup a 2 node distributed kafka-connect process. Everything went well 
> and I could see lot of data flowing into the relevant kafka topics.
> After some time, JDBCUtils.getCurrentTimeOnDB threw a ConnectException with 
> the following stacktrace:
> The last packet successfully received from the server was 792 milliseconds 
> ago.  The last packet sent successfully to the server was 286 milliseconds 
> ago. (io.confluent.connect.jdbc.source.JdbcSourceTask:234)
> [2016-11-02 12:42:06,116] ERROR Failed to get current time from DB using 
> query select CURRENT_TIMESTAMP; on database MySQL 
> (io.confluent.connect.jdbc.util.JdbcUtils:226)
> com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link 
> failure
> The last packet successfully received from the server was 1,855 milliseconds 
> ago.  The last packet sent successfully to the server was 557 milliseconds 
> ago.
>at sun.reflect.GeneratedConstructorAccessor51.newInstance(Unknown 
> Source)
>at 
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
>at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
>at com.mysql.jdbc.Util.handleNewInstance(Util.java:411)
>at 
> com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:1117)
>at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3829)
>at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2449)
>at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2629)
>at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2719)
>at 
> com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:2155)
>at 
> com.mysql.jdbc.PreparedStatement.execute(PreparedStatement.java:1379)
>at 
> com.mysql.jdbc.StatementImpl.createResultSetUsingServerFetch(StatementImpl.java:651)
>at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1527)
>at 
> io.confluent.connect.jdbc.util.JdbcUtils.getCurrentTimeOnDB(JdbcUtils.java:220)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.executeQuery(TimestampIncrementingTableQuerier.java:157)
>at 
> io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:78)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:57)
>at 
> io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:207)
>at 
> org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:155)
>at 
> org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
>at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
>at 
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
>at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
>at java.lang.Thread.run(Thread.java:745)
> Caused by: java.net.SocketException: Broken pipe (Write failed)
>at java.net.SocketOutputStream.socketWrite0(Native Method)
>at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:109)
>at java.net.SocketOutputStream.write(SocketOutputStream.java:153)
>at 
> java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
>at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
>at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3810)
>... 20 more
> This was just a minor glitch to the connection as the ec2 isntances are able 
> to connect to the Mysql Aurora instances without any issues.
> But, after this exception(which is there a number of times), none of the 
> connectors' tasks are executing. Beyond this, all I see in the logs is 
> [2016-11-02 16:17:41,983] ERROR Failed to run query for table 
> TimestampIncrementingTableQuerier{name='eng_match_series', query='null', 
> topicPrefix='ci-eng-', 

Re: [DISCUSS] KIP-84: Support SASL/SCRAM mechanisms

[Ismael Juma]

Hi all,

I'm currently with limited internet access so I have not done a proper
review of the KIP, my apologies. A couple of thoughts:

1. I agree with Rajini and I don't see the value in bumping the protocol
version in this case. As Rajini said, the enabled mechanisms are
independent of the broker version. If and when we have a feature API, we
can consider exposing it that way.

2. Do we really want to support sha1? The security of it has been
questioned and the general recommendation is to avoid it these days.

Ismael

On 3 Nov 2016 6:51 am, "Rajini Sivaram" 
wrote:

> I don't have a strong objection to bumping up SaslHandshakeRequest version,
> though I am still not convinced of the usefulness. I do agree that KIP-35
> should be standard way to determine client/broker compatibility. But I am
> not sure ApiVersionsRequest is a good fit for clients to choose a SASL
> mechanism.
>
>1. SaslHandshakeRequest is the only way a client can figure out if a
>SASL mechanism is actually enabled in the broker. The fact that broker
>version n supports SCRAM doesn't imply that a broker of version n
> actually
>has the mechanism enabled. Since enabling a mechanism involves some
> effort
>like installing an authentication server and/or configuring credentials,
>SASL mechanisms are a feature of a broker installation very unlike
>versions.  As you said, "features" would have worked better.
>2. New SASL mechanisms can be added to older Kafka broker versions. With
>some tweaking, the SCRAM implementation from KIP-84 can be enabled in a
>0.10.0 broker without changing any broker code. KIP-86 would make this
> even
>easier, but it is already possible to add new or custom mechanisms to
>existing broker versions. A client using ApiVersionsRequest to choose a
>SASL mechanism is only checking when a mechanism was included in a
> broker
>release, not the "capability" of a broker to support the mechanism. I am
>not sure we should encourage clients to choose mechanisms based on
> versions.
>3. Clients need additional configuration based on the chosen mechanism.
>One of the reasons I couldn't see any value in using ApiVersionsRequest
> in
>the Java client was because clients are configured with a single SASL
>mechanism and a JAAS configuration corresponding to that mechanism. If a
>client wants to choose between Kerberos and SCRAM, the client would need
>keytab/principal for kerberos and username/password for SCRAM. Clients
> that
>possess multiple credentials without knowing what the broker supports -
>that sounds like a rather unusual scenario.
>4. For some mechanisms, clients may want to choose mechanism at runtime.
>For example, a client is configured with username/ password and wants to
>choose between SCRAM-SHA-1/SCRAM-SHA-256/PLAIN depending on which
>mechanisms are enabled in the broker. To choose between SCRAM-SHA-1 and
>SCRAM-SHA-256, clients have to use SaslHandshakeRequest since they were
>added in the same version. To choose between PLAIN and SCRAM-SHA-256, a
>versions based approach would work, but wouldn't it be better for
> clients
>to rely on just SaslHandshakeRequest rather than
>ApiVersionsRequest+SaslHandshakeRequest so that the solution works in
> all
>scenarios?
>
>
>
>
> On Thu, Nov 3, 2016 at 4:33 AM, Ewen Cheslack-Postava 
> wrote:
>
> > I think the bump isn't strictly required, but if the client is KIP-35
> > aware, it can proactively choose a compatible SASL mechanism based on its
> > initial ApiVersionRequest and avoid an extra connection round trip when
> > there are client/broker version differences. Without this, a newer client
> > would have to do 2 set of requests since the first SASL mechanism might
> not
> > be compatible.
> >
> > I don't think this is a deal breaker, but I do think it would be good to
> > just standardize on KIP-35 as the way we figure out client/broker
> > compatibility. The SASL stuff happened in parallel (maybe before?) KIP-35
> > and ended up with its own mechanism, but I'm in favor of trying to
> simplify
> > everything by centralizing those considerations into a single API call.
> (By
> > the way, dredging up now ancient history in the KIP-35 discussion, this
> is
> > also why "features" vs "API version" is relevant. If we wanted to
> configure
> > a newer broker to disable SASL mechanisms we no longer want to allow use
> > of, this isn't really possible to express via API versions unless we also
> > explicitly add an API version that doesn't support that mechanism whereas
> > features would make this easier to toggle on/off. The
> SaslHandshakeRequest
> > probably makes it easier to keep thing secure compared to the current
> state
> > of ApiVersionRequest).
> >
> > -Ewen
> >
> > On Tue, Nov 1, 2016 at 2:09 PM, Rajini Sivaram <
> > rajinisiva...@googlemail.com
> > > wrote:
> >
> > > Gwen,
> > >
> 

[jira] [Comment Edited] (KAFKA-4371) Sporadic ConnectException shuts down the whole connect process

[Tianji Li (JIRA)]

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

Tianji Li edited comment on KAFKA-4371 at 11/3/16 11:50 AM:


This seems like a bug in the https://github.com/confluentinc/kafka-connect-jdbc 
repository, rather than Kafka itself.

The SocketException (or rather IOException) at line 
https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java#L211
 is not caught.


was (Author: skyahead):
This seems like a bug in the https://github.com/confluentinc/kafka-connect-jdbc 
repository, rather than Kafka itself.

The SQLException (or rather IOException) at line 
https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java#L211
 is not caught.

> Sporadic ConnectException shuts down the whole connect process
> --
>
> Key: KAFKA-4371
> URL: https://issues.apache.org/jira/browse/KAFKA-4371
> Project: Kafka
>  Issue Type: Bug
>Reporter: Sagar Rao
>Priority: Critical
>
> I had setup a 2 node distributed kafka-connect process. Everything went well 
> and I could see lot of data flowing into the relevant kafka topics.
> After some time, JDBCUtils.getCurrentTimeOnDB threw a ConnectException with 
> the following stacktrace:
> The last packet successfully received from the server was 792 milliseconds 
> ago.  The last packet sent successfully to the server was 286 milliseconds 
> ago. (io.confluent.connect.jdbc.source.JdbcSourceTask:234)
> [2016-11-02 12:42:06,116] ERROR Failed to get current time from DB using 
> query select CURRENT_TIMESTAMP; on database MySQL 
> (io.confluent.connect.jdbc.util.JdbcUtils:226)
> com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link 
> failure
> The last packet successfully received from the server was 1,855 milliseconds 
> ago.  The last packet sent successfully to the server was 557 milliseconds 
> ago.
>at sun.reflect.GeneratedConstructorAccessor51.newInstance(Unknown 
> Source)
>at 
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
>at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
>at com.mysql.jdbc.Util.handleNewInstance(Util.java:411)
>at 
> com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:1117)
>at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3829)
>at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2449)
>at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2629)
>at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2719)
>at 
> com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:2155)
>at 
> com.mysql.jdbc.PreparedStatement.execute(PreparedStatement.java:1379)
>at 
> com.mysql.jdbc.StatementImpl.createResultSetUsingServerFetch(StatementImpl.java:651)
>at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1527)
>at 
> io.confluent.connect.jdbc.util.JdbcUtils.getCurrentTimeOnDB(JdbcUtils.java:220)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.executeQuery(TimestampIncrementingTableQuerier.java:157)
>at 
> io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:78)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:57)
>at 
> io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:207)
>at 
> org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:155)
>at 
> org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
>at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
>at 
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
>at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
>at java.lang.Thread.run(Thread.java:745)
> Caused by: java.net.SocketException: Broken pipe (Write failed)
>at java.net.SocketOutputStream.socketWrite0(Native Method)
>at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:109)
>at java.net.SocketOutputStream.write(SocketOutputStream.java:153)
>at 
> java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
>at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
>at 

[jira] [Commented] (KAFKA-4371) Sporadic ConnectException shuts down the whole connect process

[Tianji Li (JIRA)]

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

Tianji Li commented on KAFKA-4371:
--

This seems like a bug in the https://github.com/confluentinc/kafka-connect-jdbc 
repository, rather than Kafka itself.

The SQLException (or rather IOException) at line 
https://github.com/confluentinc/kafka-connect-jdbc/blob/master/src/main/java/io/confluent/connect/jdbc/source/JdbcSourceTask.java#L211
 is not caught.

> Sporadic ConnectException shuts down the whole connect process
> --
>
> Key: KAFKA-4371
> URL: https://issues.apache.org/jira/browse/KAFKA-4371
> Project: Kafka
>  Issue Type: Bug
>Reporter: Sagar Rao
>Priority: Critical
>
> I had setup a 2 node distributed kafka-connect process. Everything went well 
> and I could see lot of data flowing into the relevant kafka topics.
> After some time, JDBCUtils.getCurrentTimeOnDB threw a ConnectException with 
> the following stacktrace:
> The last packet successfully received from the server was 792 milliseconds 
> ago.  The last packet sent successfully to the server was 286 milliseconds 
> ago. (io.confluent.connect.jdbc.source.JdbcSourceTask:234)
> [2016-11-02 12:42:06,116] ERROR Failed to get current time from DB using 
> query select CURRENT_TIMESTAMP; on database MySQL 
> (io.confluent.connect.jdbc.util.JdbcUtils:226)
> com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link 
> failure
> The last packet successfully received from the server was 1,855 milliseconds 
> ago.  The last packet sent successfully to the server was 557 milliseconds 
> ago.
>at sun.reflect.GeneratedConstructorAccessor51.newInstance(Unknown 
> Source)
>at 
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
>at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
>at com.mysql.jdbc.Util.handleNewInstance(Util.java:411)
>at 
> com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:1117)
>at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3829)
>at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2449)
>at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2629)
>at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2719)
>at 
> com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:2155)
>at 
> com.mysql.jdbc.PreparedStatement.execute(PreparedStatement.java:1379)
>at 
> com.mysql.jdbc.StatementImpl.createResultSetUsingServerFetch(StatementImpl.java:651)
>at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1527)
>at 
> io.confluent.connect.jdbc.util.JdbcUtils.getCurrentTimeOnDB(JdbcUtils.java:220)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.executeQuery(TimestampIncrementingTableQuerier.java:157)
>at 
> io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:78)
>at 
> io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:57)
>at 
> io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:207)
>at 
> org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:155)
>at 
> org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
>at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
>at 
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
>at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
>at java.lang.Thread.run(Thread.java:745)
> Caused by: java.net.SocketException: Broken pipe (Write failed)
>at java.net.SocketOutputStream.socketWrite0(Native Method)
>at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:109)
>at java.net.SocketOutputStream.write(SocketOutputStream.java:153)
>at 
> java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
>at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
>at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3810)
>... 20 more
> This was just a minor glitch to the connection as the ec2 isntances are able 
> to connect to the Mysql Aurora instances without any issues.
> But, after this exception(which is there a number of times), none of the 
> connectors' tasks are executing. Beyond this, all I see in the logs is 
> [2016-11-02 16:17:41,983] ERROR Failed to run query for table 

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

[Apache Jenkins Server]

See 

Changes:

[ismael] MINOR: Remove stray `%s` in `TopicMetadata.toString`

--
[...truncated 45972 lines...]
Caused by:
java.nio.file.FileSystemException: /tmp/kafka-4894287765309853019: No 
space left on device
at 
sun.nio.fs.UnixException.translateToIOException(UnixException.java:91)
at 
sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
at 
sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
at 
sun.nio.fs.UnixFileSystemProvider.createDirectory(UnixFileSystemProvider.java:384)
at java.nio.file.Files.createDirectory(Files.java:674)
at java.nio.file.TempFileHelper.create(TempFileHelper.java:136)
at 
java.nio.file.TempFileHelper.createTempDirectory(TempFileHelper.java:173)
at java.nio.file.Files.createTempDirectory(Files.java:991)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:163)
... 4 more

kafka.log.LogTest > testCleanShutdownFile STARTED

kafka.log.LogTest > testCleanShutdownFile FAILED
java.lang.RuntimeException: Failed to create a temp dir
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:165)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:139)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:149)
at kafka.utils.TestUtils$.tempDir(TestUtils.scala:74)
at kafka.log.LogTest.(LogTest.scala:36)

Caused by:
java.nio.file.FileSystemException: /tmp/kafka-9166521233024099469: No 
space left on device
at 
sun.nio.fs.UnixException.translateToIOException(UnixException.java:91)
at 
sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
at 
sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
at 
sun.nio.fs.UnixFileSystemProvider.createDirectory(UnixFileSystemProvider.java:384)
at java.nio.file.Files.createDirectory(Files.java:674)
at java.nio.file.TempFileHelper.create(TempFileHelper.java:136)
at 
java.nio.file.TempFileHelper.createTempDirectory(TempFileHelper.java:173)
at java.nio.file.Files.createTempDirectory(Files.java:991)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:163)
... 4 more

kafka.log.LogTest > testBuildTimeIndexWhenNotAssigningOffsets STARTED

kafka.log.LogTest > testBuildTimeIndexWhenNotAssigningOffsets FAILED
java.lang.RuntimeException: Failed to create a temp dir
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:165)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:139)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:149)
at kafka.utils.TestUtils$.tempDir(TestUtils.scala:74)
at kafka.log.LogTest.(LogTest.scala:36)

Caused by:
java.nio.file.FileSystemException: /tmp/kafka-9205157461024087189: No 
space left on device
at 
sun.nio.fs.UnixException.translateToIOException(UnixException.java:91)
at 
sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
at 
sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
at 
sun.nio.fs.UnixFileSystemProvider.createDirectory(UnixFileSystemProvider.java:384)
at java.nio.file.Files.createDirectory(Files.java:674)
at java.nio.file.TempFileHelper.create(TempFileHelper.java:136)
at 
java.nio.file.TempFileHelper.createTempDirectory(TempFileHelper.java:173)
at java.nio.file.Files.createTempDirectory(Files.java:991)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:163)
... 4 more

kafka.log.LogConfigTest > shouldValidateThrottledReplicasConfig STARTED

kafka.log.LogConfigTest > shouldValidateThrottledReplicasConfig PASSED

kafka.log.LogConfigTest > testFromPropsEmpty STARTED

kafka.log.LogConfigTest > testFromPropsEmpty PASSED

kafka.log.LogConfigTest > testKafkaConfigToProps STARTED

kafka.log.LogConfigTest > testKafkaConfigToProps FAILED
java.lang.RuntimeException: Failed to create a temp dir
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:165)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:139)
at org.apache.kafka.test.TestUtils.tempDirectory(TestUtils.java:149)
at kafka.utils.TestUtils$.tempDir(TestUtils.scala:74)
at kafka.utils.TestUtils$.createBrokerConfig(TestUtils.scala:189)
at 
kafka.log.LogConfigTest.testKafkaConfigToProps(LogConfigTest.scala:34)

Caused by:
java.nio.file.FileSystemException: /tmp/kafka-3925811358755630771: No 
space left on device
at 
sun.nio.fs.UnixException.translateToIOException(UnixException.java:91)
 

[jira] [Commented] (KAFKA-2247) Merge kafka.utils.Time and kafka.common.utils.Time

[Ismael Juma (JIRA)]

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

Ismael Juma commented on KAFKA-2247:


I submitted a PR since no progress has been made on this for a while.

> Merge kafka.utils.Time and kafka.common.utils.Time
> --
>
> Key: KAFKA-2247
> URL: https://issues.apache.org/jira/browse/KAFKA-2247
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Aditya Auradkar
>Assignee: Ismael Juma
>Priority: Minor
> Fix For: 0.10.2.0
>
>
> We currently have 2 different versions of Time in clients and core. These 
> need to be merged.
> It's worth noting that `kafka.utils.MockTime` includes a `scheduler` that is 
> used by some tests while `o.a.kafka.common.utils.Time` does not. We either 
> need to add this functionality or change the tests not to need it anymore.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2091: MINOR: Bug fixed

[asfgit]

Github user asfgit closed the pull request at:

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


---
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-2247) Merge kafka.utils.Time and kafka.common.utils.Time

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-2247:
---
Fix Version/s: 0.10.2.0
   Status: Patch Available  (was: Open)

> Merge kafka.utils.Time and kafka.common.utils.Time
> --
>
> Key: KAFKA-2247
> URL: https://issues.apache.org/jira/browse/KAFKA-2247
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Aditya Auradkar
>Assignee: Ismael Juma
>Priority: Minor
> Fix For: 0.10.2.0
>
>
> We currently have 2 different versions of Time in clients and core. These 
> need to be merged.
> It's worth noting that `kafka.utils.MockTime` includes a `scheduler` that is 
> used by some tests while `o.a.kafka.common.utils.Time` does not. We either 
> need to add this functionality or change the tests not to need it anymore.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-2247) Merge kafka.utils.Time and kafka.common.utils.Time

[ASF GitHub Bot (JIRA)]

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

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

GitHub user ijuma opened a pull request:

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

KAFKA-2247: Merge kafka.utils.Time and kafka.common.utils.Time



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

$ git pull https://github.com/ijuma/kafka 
kafka-2247-consolidate-time-interfaces

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

https://github.com/apache/kafka/pull/2095.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 #2095


commit 5126636282ca234ba9945acf9959d3182b20b33c
Author: Ismael Juma 
Date:   2016-11-03T10:17:54Z

KAFKA-2247: Merge kafka.utils.Time and kafka.common.utils.Time




> Merge kafka.utils.Time and kafka.common.utils.Time
> --
>
> Key: KAFKA-2247
> URL: https://issues.apache.org/jira/browse/KAFKA-2247
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Aditya Auradkar
>Assignee: Ismael Juma
>Priority: Minor
>
> We currently have 2 different versions of Time in clients and core. These 
> need to be merged.
> It's worth noting that `kafka.utils.MockTime` includes a `scheduler` that is 
> used by some tests while `o.a.kafka.common.utils.Time` does not. We either 
> need to add this functionality or change the tests not to need it anymore.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[GitHub] kafka pull request #2095: KAFKA-2247: Merge kafka.utils.Time and kafka.commo...

[ijuma]

GitHub user ijuma opened a pull request:

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

KAFKA-2247: Merge kafka.utils.Time and kafka.common.utils.Time



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

$ git pull https://github.com/ijuma/kafka 
kafka-2247-consolidate-time-interfaces

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

https://github.com/apache/kafka/pull/2095.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 #2095


commit 5126636282ca234ba9945acf9959d3182b20b33c
Author: Ismael Juma 
Date:   2016-11-03T10:17:54Z

KAFKA-2247: Merge kafka.utils.Time and kafka.common.utils.Time




---
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-84: Support SASL/SCRAM mechanisms

[Rajini Sivaram]

I don't have a strong objection to bumping up SaslHandshakeRequest version,
though I am still not convinced of the usefulness. I do agree that KIP-35
should be standard way to determine client/broker compatibility. But I am
not sure ApiVersionsRequest is a good fit for clients to choose a SASL
mechanism.

   1. SaslHandshakeRequest is the only way a client can figure out if a
   SASL mechanism is actually enabled in the broker. The fact that broker
   version n supports SCRAM doesn't imply that a broker of version n actually
   has the mechanism enabled. Since enabling a mechanism involves some effort
   like installing an authentication server and/or configuring credentials,
   SASL mechanisms are a feature of a broker installation very unlike
   versions.  As you said, "features" would have worked better.
   2. New SASL mechanisms can be added to older Kafka broker versions. With
   some tweaking, the SCRAM implementation from KIP-84 can be enabled in a
   0.10.0 broker without changing any broker code. KIP-86 would make this even
   easier, but it is already possible to add new or custom mechanisms to
   existing broker versions. A client using ApiVersionsRequest to choose a
   SASL mechanism is only checking when a mechanism was included in a broker
   release, not the "capability" of a broker to support the mechanism. I am
   not sure we should encourage clients to choose mechanisms based on versions.
   3. Clients need additional configuration based on the chosen mechanism.
   One of the reasons I couldn't see any value in using ApiVersionsRequest in
   the Java client was because clients are configured with a single SASL
   mechanism and a JAAS configuration corresponding to that mechanism. If a
   client wants to choose between Kerberos and SCRAM, the client would need
   keytab/principal for kerberos and username/password for SCRAM. Clients that
   possess multiple credentials without knowing what the broker supports -
   that sounds like a rather unusual scenario.
   4. For some mechanisms, clients may want to choose mechanism at runtime.
   For example, a client is configured with username/ password and wants to
   choose between SCRAM-SHA-1/SCRAM-SHA-256/PLAIN depending on which
   mechanisms are enabled in the broker. To choose between SCRAM-SHA-1 and
   SCRAM-SHA-256, clients have to use SaslHandshakeRequest since they were
   added in the same version. To choose between PLAIN and SCRAM-SHA-256, a
   versions based approach would work, but wouldn't it be better for clients
   to rely on just SaslHandshakeRequest rather than
   ApiVersionsRequest+SaslHandshakeRequest so that the solution works in all
   scenarios?




On Thu, Nov 3, 2016 at 4:33 AM, Ewen Cheslack-Postava 
wrote:

> I think the bump isn't strictly required, but if the client is KIP-35
> aware, it can proactively choose a compatible SASL mechanism based on its
> initial ApiVersionRequest and avoid an extra connection round trip when
> there are client/broker version differences. Without this, a newer client
> would have to do 2 set of requests since the first SASL mechanism might not
> be compatible.
>
> I don't think this is a deal breaker, but I do think it would be good to
> just standardize on KIP-35 as the way we figure out client/broker
> compatibility. The SASL stuff happened in parallel (maybe before?) KIP-35
> and ended up with its own mechanism, but I'm in favor of trying to simplify
> everything by centralizing those considerations into a single API call. (By
> the way, dredging up now ancient history in the KIP-35 discussion, this is
> also why "features" vs "API version" is relevant. If we wanted to configure
> a newer broker to disable SASL mechanisms we no longer want to allow use
> of, this isn't really possible to express via API versions unless we also
> explicitly add an API version that doesn't support that mechanism whereas
> features would make this easier to toggle on/off. The SaslHandshakeRequest
> probably makes it easier to keep thing secure compared to the current state
> of ApiVersionRequest).
>
> -Ewen
>
> On Tue, Nov 1, 2016 at 2:09 PM, Rajini Sivaram <
> rajinisiva...@googlemail.com
> > wrote:
>
> > Gwen,
> >
> > I had thought the same too and hence I am assuming that Java clients
> could
> > simply use SaslHandshakeRequest. SaslHandshakeRequest returns the list of
> > mechanisms enabled in the broker. I think Jun's point was that by
> > incrementing the version of SaslHandshakeRequest, clients can use
> > ApiVersionsRequest to figure out the mechanisms the broker is capable of
> > supporting and use that information to choose a mechanism to send in
> > SaslHandshakeRequest. Not sure how useful this actually is, so will wait
> > for Jun's response.
> >
> >
> >
> > On Tue, Nov 1, 2016 at 8:18 PM, Gwen Shapira  wrote:
> >
> > > Wait, I thought SaslHandshakeResponse includes a list of mechanisms
> > > supported, so I'm not sure why we need to bump the version?
> 

[jira] [Assigned] (KAFKA-2247) Merge kafka.utils.Time and kafka.common.utils.Time

[Ismael Juma (JIRA)]

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

Ismael Juma reassigned KAFKA-2247:
--

Assignee: Ismael Juma  (was: Aditya Auradkar)

> Merge kafka.utils.Time and kafka.common.utils.Time
> --
>
> Key: KAFKA-2247
> URL: https://issues.apache.org/jira/browse/KAFKA-2247
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Aditya Auradkar
>Assignee: Ismael Juma
>Priority: Minor
>
> We currently have 2 different versions of Time in clients and core. These 
> need to be merged.
> It's worth noting that `kafka.utils.MockTime` includes a `scheduler` that is 
> used by some tests while `o.a.kafka.common.utils.Time` does not. We either 
> need to add this functionality or change the tests not to need it anymore.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4371) Sporadic ConnectException shuts down the whole connect process

[Sagar Rao (JIRA)]

Sagar Rao created KAFKA-4371:


 Summary: Sporadic ConnectException shuts down the whole connect 
process
 Key: KAFKA-4371
 URL: https://issues.apache.org/jira/browse/KAFKA-4371
 Project: Kafka
  Issue Type: Bug
Reporter: Sagar Rao
Priority: Critical


I had setup a 2 node distributed kafka-connect process. Everything went well 
and I could see lot of data flowing into the relevant kafka topics.

After some time, JDBCUtils.getCurrentTimeOnDB threw a ConnectException with the 
following stacktrace:

The last packet successfully received from the server was 792 milliseconds ago. 
 The last packet sent successfully to the server was 286 milliseconds ago. 
(io.confluent.connect.jdbc.source.JdbcSourceTask:234)
[2016-11-02 12:42:06,116] ERROR Failed to get current time from DB using query 
select CURRENT_TIMESTAMP; on database MySQL 
(io.confluent.connect.jdbc.util.JdbcUtils:226)
com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link 
failure

The last packet successfully received from the server was 1,855 milliseconds 
ago.  The last packet sent successfully to the server was 557 milliseconds ago.
   at sun.reflect.GeneratedConstructorAccessor51.newInstance(Unknown Source)
   at 
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
   at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
   at com.mysql.jdbc.Util.handleNewInstance(Util.java:411)
   at 
com.mysql.jdbc.SQLError.createCommunicationsException(SQLError.java:1117)
   at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3829)
   at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2449)
   at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2629)
   at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2719)
   at 
com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:2155)
   at com.mysql.jdbc.PreparedStatement.execute(PreparedStatement.java:1379)
   at 
com.mysql.jdbc.StatementImpl.createResultSetUsingServerFetch(StatementImpl.java:651)
   at com.mysql.jdbc.StatementImpl.executeQuery(StatementImpl.java:1527)
   at 
io.confluent.connect.jdbc.util.JdbcUtils.getCurrentTimeOnDB(JdbcUtils.java:220)
   at 
io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.executeQuery(TimestampIncrementingTableQuerier.java:157)
   at 
io.confluent.connect.jdbc.source.TableQuerier.maybeStartQuery(TableQuerier.java:78)
   at 
io.confluent.connect.jdbc.source.TimestampIncrementingTableQuerier.maybeStartQuery(TimestampIncrementingTableQuerier.java:57)
   at 
io.confluent.connect.jdbc.source.JdbcSourceTask.poll(JdbcSourceTask.java:207)
   at 
org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:155)
   at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:140)
   at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:175)
   at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
   at java.util.concurrent.FutureTask.run(FutureTask.java:266)
   at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
   at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
   at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.SocketException: Broken pipe (Write failed)
   at java.net.SocketOutputStream.socketWrite0(Native Method)
   at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:109)
   at java.net.SocketOutputStream.write(SocketOutputStream.java:153)
   at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
   at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140)
   at com.mysql.jdbc.MysqlIO.send(MysqlIO.java:3810)
   ... 20 more

This was just a minor glitch to the connection as the ec2 isntances are able to 
connect to the Mysql Aurora instances without any issues.

But, after this exception(which is there a number of times), none of the 
connectors' tasks are executing. Beyond this, all I see in the logs is 

[2016-11-02 16:17:41,983] ERROR Failed to run query for table 
TimestampIncrementingTableQuerier{name='eng_match_series', query='null', 
topicPrefix='ci-eng-', timestampColumn='modified', incrementingColumn='id'}: 
com.mysql.jdbc.exceptions.jdbc4.MySQLNonTransientConnectionException: No 
operations allowed after statement closed. 
(io.confluent.connect.jdbc.source.JdbcSourceTask:234)

Is this expected behaviour? I restarted the connector using REST apis but that 
didn't help. How do we handle such scenarios? 

Eventually I had to delete the connector and restart.

The kafka version I am using is 0.10.0.1-cp1 as there were some custom changes 
we needed to make at the Connect level.




--
This message was sent by Atlassian JIRA

[jira] [Commented] (KAFKA-4367) MirrorMaker shuts down gracefully without actually being stopped

[Alex (JIRA)]

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

Alex commented on KAFKA-4367:
-

This shouldn't be an issue since we run MirrorMaker detached from terminal: &> 
/dev/null & 
Yes, it does look like a clean shutdown and that's wierd. 

> MirrorMaker shuts down gracefully without actually being stopped
> 
>
> Key: KAFKA-4367
> URL: https://issues.apache.org/jira/browse/KAFKA-4367
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.9.0.1
> Environment: RHEL 7
>Reporter: Alex
>
> MirrorMaker stops working without being stopped. From 30 minutes to 20 hours. 
> No clue why this problem occurs.
> Start:
> bin/kafka-mirror-maker.sh --new.consumer --consumer.config 
> config/ssl_mirroring_consumer.properties --producer.config 
> config/ssl_mirroring_producer.properties --whitelist 
> "TOPIC1|TOPIC2|TOPIC3|TOPIC4" --num.streams 20 &> /dev/null &
> 
>   kafka-mirror-maker.log
> 
> [2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
> CEP.FS.IN-175 with base offset offset 15015 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
> CEP.FS.IN-151 with base offset offset 15066 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,003] TRACE Nodes with data ready to send: [Node(8, 
> 10.126.0.2, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,003] TRACE Created 1 produce requests: 
> [ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.producer.internals.Sender$1@483c4c7a, 
> request=RequestSend(header={api_key=0,api_version=1,correlation_id=219685,client_id=producer-1},
>  
> body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=133,record_set=java.nio.HeapByteBuffer[pos=0
>  lim=9085 cap=16384]}]}]}), createdTimeMs=1478017412003, sendTimeMs=0)] 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,008] TRACE Returning fetched records for assigned 
> partition CEP.FS.IN-172 and update consumed position to 3869316 
> (org.apache.kafka.clients.consumer.internals.Fetcher)
> [2016-11-01 19:23:32,008] TRACE [mirrormaker-thread-7] Sending message with 
> value size 485 and offset 3869315 (kafka.tools.MirrorMaker$MirrorMakerThread)
> [2016-11-01 19:23:32,008] TRACE Sending record 
> ProducerRecord(topic=CEP.FS.IN, partition=null, key=null, value=[B@12a54f5a 
> with callback kafka.tools.MirrorMaker$MirrorMakerProducerCallback@5ea65b8f to 
> topic CEP.FS.IN partition 160 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2016-11-01 19:23:32,008] TRACE Allocating a new 16384 byte message buffer 
> for topic CEP.FS.IN partition 160 
> (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2016-11-01 19:23:32,008] TRACE Waking up the sender since topic CEP.FS.IN 
> partition 160 is either full or getting a new batch 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2016-11-01 19:23:32,010] TRACE Received produce response from node 7 with 
> correlation id 219684 (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
> CEP.FS.IN-106 with base offset offset 15086 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
> CEP.FS.IN-124 with base offset offset 15095 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,010] TRACE Nodes with data ready to send: [Node(7, 
> 10.126.0.1, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] INFO Start clean shutdown. 
> (kafka.tools.MirrorMaker$)
> [2016-11-01 19:23:32,010] TRACE Created 1 produce requests: 
> [ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.producer.internals.Sender$1@44b788c7, 
> request=RequestSend(header={api_key=0,api_version=1,correlation_id=219686,client_id=producer-1},
>  
> body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=160,record_set=java.nio.HeapByteBuffer[pos=0
>  lim=511 cap=16384]}]}]}), createdTimeMs=1478017412010, sendTimeMs=0)] 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] INFO Shutting down consumer threads. 
> (kafka.tools.MirrorMaker$)
> [2016-11-01 19:23:32,011] INFO [mirrormaker-thread-0] mirrormaker-thread-0 
> shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
> 

[jira] [Updated] (KAFKA-4367) MirrorMaker shuts down gracefully without actually being stopped

[Alex (JIRA)]

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

Alex updated KAFKA-4367:

Description: 
MirrorMaker stops working without being stopped. From 30 minutes to 20 hours. 
No clue why this problem occurs.

Start:
bin/kafka-mirror-maker.sh --new.consumer --consumer.config 
config/ssl_mirroring_consumer.properties --producer.config 
config/ssl_mirroring_producer.properties --whitelist 
"TOPIC1|TOPIC2|TOPIC3|TOPIC4" --num.streams 20 &> /dev/null &


  kafka-mirror-maker.log

[2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
CEP.FS.IN-175 with base offset offset 15015 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
CEP.FS.IN-151 with base offset offset 15066 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,003] TRACE Nodes with data ready to send: [Node(8, 
10.126.0.2, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,003] TRACE Created 1 produce requests: 
[ClientRequest(expectResponse=true, 
callback=org.apache.kafka.clients.producer.internals.Sender$1@483c4c7a, 
request=RequestSend(header={api_key=0,api_version=1,correlation_id=219685,client_id=producer-1},
 
body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=133,record_set=java.nio.HeapByteBuffer[pos=0
 lim=9085 cap=16384]}]}]}), createdTimeMs=1478017412003, sendTimeMs=0)] 
(org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,008] TRACE Returning fetched records for assigned 
partition CEP.FS.IN-172 and update consumed position to 3869316 
(org.apache.kafka.clients.consumer.internals.Fetcher)
[2016-11-01 19:23:32,008] TRACE [mirrormaker-thread-7] Sending message with 
value size 485 and offset 3869315 (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,008] TRACE Sending record ProducerRecord(topic=CEP.FS.IN, 
partition=null, key=null, value=[B@12a54f5a with callback 
kafka.tools.MirrorMaker$MirrorMakerProducerCallback@5ea65b8f to topic CEP.FS.IN 
partition 160 (org.apache.kafka.clients.producer.KafkaProducer)
[2016-11-01 19:23:32,008] TRACE Allocating a new 16384 byte message buffer for 
topic CEP.FS.IN partition 160 
(org.apache.kafka.clients.producer.internals.RecordAccumulator)
[2016-11-01 19:23:32,008] TRACE Waking up the sender since topic CEP.FS.IN 
partition 160 is either full or getting a new batch 
(org.apache.kafka.clients.producer.KafkaProducer)
[2016-11-01 19:23:32,010] TRACE Received produce response from node 7 with 
correlation id 219684 (org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
CEP.FS.IN-106 with base offset offset 15086 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
CEP.FS.IN-124 with base offset offset 15095 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,010] TRACE Nodes with data ready to send: [Node(7, 
10.126.0.1, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,010] INFO Start clean shutdown. (kafka.tools.MirrorMaker$)
[2016-11-01 19:23:32,010] TRACE Created 1 produce requests: 
[ClientRequest(expectResponse=true, 
callback=org.apache.kafka.clients.producer.internals.Sender$1@44b788c7, 
request=RequestSend(header={api_key=0,api_version=1,correlation_id=219686,client_id=producer-1},
 
body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=160,record_set=java.nio.HeapByteBuffer[pos=0
 lim=511 cap=16384]}]}]}), createdTimeMs=1478017412010, sendTimeMs=0)] 
(org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,010] INFO Shutting down consumer threads. 
(kafka.tools.MirrorMaker$)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-0] mirrormaker-thread-0 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-1] mirrormaker-thread-1 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-2] mirrormaker-thread-2 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-3] mirrormaker-thread-3 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-4] mirrormaker-thread-4 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-5] mirrormaker-thread-5 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,012] INFO [mirrormaker-thread-6] 

[jira] [Updated] (KAFKA-4367) MirrorMaker shuts down gracefully without actually being stopped

[Alex (JIRA)]

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

Alex updated KAFKA-4367:

Description: 
MirrorMaker stops working without being stopped, 30 minutes after start. No 
clue why this problem occurs.

Start:
bin/kafka-mirror-maker.sh --new.consumer --consumer.config 
config/ssl_mirroring_consumer.properties --producer.config 
config/ssl_mirroring_producer.properties --whitelist 
"TOPIC1|TOPIC2|TOPIC3|TOPIC4" --num.streams 20 &> /dev/null &


  kafka-mirror-maker.log

[2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
CEP.FS.IN-175 with base offset offset 15015 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
CEP.FS.IN-151 with base offset offset 15066 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,003] TRACE Nodes with data ready to send: [Node(8, 
10.126.0.2, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,003] TRACE Created 1 produce requests: 
[ClientRequest(expectResponse=true, 
callback=org.apache.kafka.clients.producer.internals.Sender$1@483c4c7a, 
request=RequestSend(header={api_key=0,api_version=1,correlation_id=219685,client_id=producer-1},
 
body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=133,record_set=java.nio.HeapByteBuffer[pos=0
 lim=9085 cap=16384]}]}]}), createdTimeMs=1478017412003, sendTimeMs=0)] 
(org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,008] TRACE Returning fetched records for assigned 
partition CEP.FS.IN-172 and update consumed position to 3869316 
(org.apache.kafka.clients.consumer.internals.Fetcher)
[2016-11-01 19:23:32,008] TRACE [mirrormaker-thread-7] Sending message with 
value size 485 and offset 3869315 (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,008] TRACE Sending record ProducerRecord(topic=CEP.FS.IN, 
partition=null, key=null, value=[B@12a54f5a with callback 
kafka.tools.MirrorMaker$MirrorMakerProducerCallback@5ea65b8f to topic CEP.FS.IN 
partition 160 (org.apache.kafka.clients.producer.KafkaProducer)
[2016-11-01 19:23:32,008] TRACE Allocating a new 16384 byte message buffer for 
topic CEP.FS.IN partition 160 
(org.apache.kafka.clients.producer.internals.RecordAccumulator)
[2016-11-01 19:23:32,008] TRACE Waking up the sender since topic CEP.FS.IN 
partition 160 is either full or getting a new batch 
(org.apache.kafka.clients.producer.KafkaProducer)
[2016-11-01 19:23:32,010] TRACE Received produce response from node 7 with 
correlation id 219684 (org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
CEP.FS.IN-106 with base offset offset 15086 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
CEP.FS.IN-124 with base offset offset 15095 and error: null. 
(org.apache.kafka.clients.producer.internals.RecordBatch)
[2016-11-01 19:23:32,010] TRACE Nodes with data ready to send: [Node(7, 
10.126.0.1, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,010] INFO Start clean shutdown. (kafka.tools.MirrorMaker$)
[2016-11-01 19:23:32,010] TRACE Created 1 produce requests: 
[ClientRequest(expectResponse=true, 
callback=org.apache.kafka.clients.producer.internals.Sender$1@44b788c7, 
request=RequestSend(header={api_key=0,api_version=1,correlation_id=219686,client_id=producer-1},
 
body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=160,record_set=java.nio.HeapByteBuffer[pos=0
 lim=511 cap=16384]}]}]}), createdTimeMs=1478017412010, sendTimeMs=0)] 
(org.apache.kafka.clients.producer.internals.Sender)
[2016-11-01 19:23:32,010] INFO Shutting down consumer threads. 
(kafka.tools.MirrorMaker$)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-0] mirrormaker-thread-0 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-1] mirrormaker-thread-1 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-2] mirrormaker-thread-2 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-3] mirrormaker-thread-3 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-4] mirrormaker-thread-4 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,011] INFO [mirrormaker-thread-5] mirrormaker-thread-5 
shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
[2016-11-01 19:23:32,012] INFO [mirrormaker-thread-6] 

[jira] [Updated] (KAFKA-4367) MirrorMaker shuts down gracefully without actually being stopped

[Alex (JIRA)]

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

Alex updated KAFKA-4367:

Summary: MirrorMaker shuts down gracefully without actually being stopped  
(was: MirrorMaker shuts down gracefully without being stopped)

> MirrorMaker shuts down gracefully without actually being stopped
> 
>
> Key: KAFKA-4367
> URL: https://issues.apache.org/jira/browse/KAFKA-4367
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.9.0.1
> Environment: RHEL 7
>Reporter: Alex
>
> Start:
> bin/kafka-mirror-maker.sh --new.consumer --consumer.config 
> config/ssl_mirroring_consumer.properties --producer.config 
> config/ssl_mirroring_producer.properties --whitelist 
> "TOPIC1|TOPIC2|TOPIC3|TOPIC4" --num.streams 20 &> /dev/null &
> MirrorMaker stops working without being stopped, 30 minutes after start. No 
> clue why this problem occurs.
> 
>   kafka-mirror-maker.log
> 
> [2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
> CEP.FS.IN-175 with base offset offset 15015 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,003] TRACE Produced messages to topic-partition 
> CEP.FS.IN-151 with base offset offset 15066 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,003] TRACE Nodes with data ready to send: [Node(8, 
> 10.126.0.2, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,003] TRACE Created 1 produce requests: 
> [ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.producer.internals.Sender$1@483c4c7a, 
> request=RequestSend(header={api_key=0,api_version=1,correlation_id=219685,client_id=producer-1},
>  
> body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=133,record_set=java.nio.HeapByteBuffer[pos=0
>  lim=9085 cap=16384]}]}]}), createdTimeMs=1478017412003, sendTimeMs=0)] 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,008] TRACE Returning fetched records for assigned 
> partition CEP.FS.IN-172 and update consumed position to 3869316 
> (org.apache.kafka.clients.consumer.internals.Fetcher)
> [2016-11-01 19:23:32,008] TRACE [mirrormaker-thread-7] Sending message with 
> value size 485 and offset 3869315 (kafka.tools.MirrorMaker$MirrorMakerThread)
> [2016-11-01 19:23:32,008] TRACE Sending record 
> ProducerRecord(topic=CEP.FS.IN, partition=null, key=null, value=[B@12a54f5a 
> with callback kafka.tools.MirrorMaker$MirrorMakerProducerCallback@5ea65b8f to 
> topic CEP.FS.IN partition 160 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2016-11-01 19:23:32,008] TRACE Allocating a new 16384 byte message buffer 
> for topic CEP.FS.IN partition 160 
> (org.apache.kafka.clients.producer.internals.RecordAccumulator)
> [2016-11-01 19:23:32,008] TRACE Waking up the sender since topic CEP.FS.IN 
> partition 160 is either full or getting a new batch 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2016-11-01 19:23:32,010] TRACE Received produce response from node 7 with 
> correlation id 219684 (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
> CEP.FS.IN-106 with base offset offset 15086 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,010] TRACE Produced messages to topic-partition 
> CEP.FS.IN-124 with base offset offset 15095 and error: null. 
> (org.apache.kafka.clients.producer.internals.RecordBatch)
> [2016-11-01 19:23:32,010] TRACE Nodes with data ready to send: [Node(7, 
> 10.126.0.1, 9092)] (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] INFO Start clean shutdown. 
> (kafka.tools.MirrorMaker$)
> [2016-11-01 19:23:32,010] TRACE Created 1 produce requests: 
> [ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.producer.internals.Sender$1@44b788c7, 
> request=RequestSend(header={api_key=0,api_version=1,correlation_id=219686,client_id=producer-1},
>  
> body={acks=-1,timeout=3,topic_data=[{topic=CEP.FS.IN,data=[{partition=160,record_set=java.nio.HeapByteBuffer[pos=0
>  lim=511 cap=16384]}]}]}), createdTimeMs=1478017412010, sendTimeMs=0)] 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-11-01 19:23:32,010] INFO Shutting down consumer threads. 
> (kafka.tools.MirrorMaker$)
> [2016-11-01 19:23:32,011] INFO [mirrormaker-thread-0] mirrormaker-thread-0 
> shutting down (kafka.tools.MirrorMaker$MirrorMakerThread)
> [2016-11-01 19:23:32,011] INFO [mirrormaker-thread-1] 

[jira] [Commented] (KAFKA-4369) ZkClient is not closed upon streams shutdown

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax commented on KAFKA-4369:


There is a PR for https://issues.apache.org/jira/browse/KAFKA-4060 already that 
should get merged soon, making this JIRA void.

> ZkClient is not closed upon streams shutdown
> 
>
> Key: KAFKA-4369
> URL: https://issues.apache.org/jira/browse/KAFKA-4369
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Ryan P
>Assignee: Guozhang Wang
>
> Kafka Stream's InternalTopicManager creates a new ZkClient but fails to close 
> it as part of it's shutdown. 
> https://github.com/confluentinc/kafka/blob/v3.0.1/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java#L93
> This is likely only an issue when performing testing/debugging where the 
> streams application is shutdown but the JVM remains in tact. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-955) After a leader change, messages sent with ack=0 are lost

[Bill Zhang (JIRA)]

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

Bill Zhang commented on KAFKA-955:
--

I am using Flume with Kafka Channel & facing below issues. 

Kafka Version: kafka_2.9.1-0.8.2.0
Flume Version: apache-flume-1.6.0

It seems was resolved from below :
Step 1: copy zookeeper Jar file to Flume classpath
Step 2: a1.channels.c1.kafka.producer.type = async

Note:
i didn't change default value of request.required.acks. It seems works, it is 
still in testing...


~~
Issue 1:
02 Nov 2016 22:20:06,201 WARN  
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-2] 
(kafka.utils.Logging$class.warn:83)  - Reconnect due to socket error: null
02 Nov 2016 22:20:06,203 INFO  
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-2] 
(kafka.utils.Logging$class.info:68)  - 
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-2], 
Stopped 
02 Nov 2016 22:20:06,203 INFO  [agent-shutdown-hook] 
(kafka.utils.Logging$class.info:68)  - 
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-2], 
Shutdown completed
02 Nov 2016 22:20:06,203 INFO  [agent-shutdown-hook] 
(kafka.utils.Logging$class.info:68)  - 
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-1], 
Shutting down
02 Nov 2016 22:20:06,204 WARN  
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-1] 
(kafka.utils.Logging$class.warn:83)  - Reconnect due to socket error: null
02 Nov 2016 22:20:06,204 INFO  
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-1] 
(kafka.utils.Logging$class.info:68)  - 
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-1], 
Stopped 
02 Nov 2016 22:20:06,204 INFO  [agent-shutdown-hook] 
(kafka.utils.Logging$class.info:68)  - 
[ConsumerFetcherThread-flume_tbox-topic_SATL2036-1478087994030-54387da2-0-1], 
Shutdown completed
02 Nov 2016 22:20:06,205 INFO  [agent-shutdown-hook] 
(kafka.utils.Logging$class.info:68)  - [ConsumerFetcherManager-1478087994042] 
All connections stopped
02 Nov 2016 22:20:06,207 INFO  
[ZkClient-EventThread-58-SATL2036:2181,SATL2037:2181,SATL2038:2181/kafka] 
(org.I0Itec.zkclient.ZkEventThread.run:82)  - Terminate ZkClient event thread.
02 Nov 2016 22:20:06,212 WARN  [PollableSourceRunner-KafkaSource-r1] 
(kafka.utils.Logging$class.warn:89)  - Failed to send producer request with 
correlation id 34198503 to broker 1 with data for partitions 
[channel-tbox-parsed-topic,3]
java.nio.channels.ClosedByInterruptException
at 
java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:202)
at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:511)
at java.nio.channels.SocketChannel.write(SocketChannel.java:502)
at 
kafka.network.BoundedByteBufferSend.writeTo(BoundedByteBufferSend.scala:56)
at kafka.network.Send$class.writeCompletely(Transmission.scala:75)
at 
kafka.network.BoundedByteBufferSend.writeCompletely(BoundedByteBufferSend.scala:26)
at kafka.network.BlockingChannel.send(BlockingChannel.scala:92)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:72)
at 
kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:71)
at 
kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SyncProducer.scala:102)
at 
kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:102)
at 
kafka.producer.SyncProducer$$anonfun$send$1$$anonfun$apply$mcV$sp$1.apply(SyncProducer.scala:102)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at 
kafka.producer.SyncProducer$$anonfun$send$1.apply$mcV$sp(SyncProducer.scala:101)
at 
kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:101)
at 
kafka.producer.SyncProducer$$anonfun$send$1.apply(SyncProducer.scala:101)
at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
at kafka.producer.SyncProducer.send(SyncProducer.scala:100)
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.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772)
at 
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at 
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:98)
at