[jira] [Commented] (KAFKA-5430) new consumers getting data for revoked partitions

2017-08-07 Thread Lior Chaga (JIRA) 

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

Lior Chaga commented on KAFKA-5430:
---

Thanks [~jasong35], this certainly seems related. We will upgrade soon.

> new consumers getting data for revoked partitions
> -
>
> Key: KAFKA-5430
> URL: https://issues.apache.org/jira/browse/KAFKA-5430
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, consumer
>Affects Versions: 0.10.2.0
>Reporter: Lior Chaga
> Attachments: consumer-thread.log, consumer-thread.log, 
> kafka_trace.log.gz
>
>
> Due to bad configuration applied to network components, we experienced issues 
> with communication between kafka brokers (causing under replication) as well 
> as producers/consumers not being able to work against kafka.
> The symptoms on the consumer were many errors of the following form:
> {code}
> 2017-06-04 04:27:35,200 ERROR [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] TaboolaKafkaConsumer [] - Failed 
> committing to kafka topicPartitions 
> [requestlegacy-2,requestlegacy-0,requestlegacy-1] 
> org.apache.kafka.clients.consumer.RetriableCommitFailedException: Offset 
> commit failed with a retriable exception. You should retry committing offsets.
> Caused by: org.apache.kafka.common.errors.DisconnectException
> {code}
> So far so good. However, upon network recovery, there were several rebalance 
> operations, which eventually resulted in only one consumer (#14) being 
> assigned with all topic partitions (at this case we're talking about a 
> consumer groups for which all consumers are running in same process):
> {code}
> 2017-06-04 04:27:02,168 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-14_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-8, requestlegacy-9] 
> for group session_parser_02
> 2017-06-04 04:27:04,208 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-15_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-10, requestlegacy-11] 
> for group session_parser_02
> 2017-06-04 04:27:18,167 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-12_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-3, requestlegacy-4, 
> requestlegacy-5] for group session_parser_02
> 2017-06-04 04:27:20,232 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-2, requestlegacy-0, 
> requestlegacy-1] for group session_parser_02
> 2017-06-04 04:27:20,236 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-15_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-9, requestlegacy-10, 
> requestlegacy-11] for group session_parser_02
> 2017-06-04 04:27:20,237 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-12_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-3, requestlegacy-4, requestlegacy-5] 
> for group session_parser_02
> 2017-06-04 04:27:20,237 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-14_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-6, requestlegacy-7, requestlegacy-8] 
> for group session_parser_02
> 2017-06-04 04:27:20,332 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-2, requestlegacy-0, requestlegacy-1] 
> for group session_parser_02
> 2017-06-04 04:28:52,368 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-13_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-6, requestlegacy-7] 
> for group session_parser_02
> 2017-06-04 04:29:15,201 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-2, requestlegacy-0, 
> requestlegacy-1] for group session_parser_02
> 2017-06-04 04:30:22,379 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-14_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-6, requestlegacy-7, 
> requestlegacy-8] for group session_parser_02
> 2017-06-04 04:30:24,431 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-15_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-9, requestlegacy-10, 
> requestlegacy-11] for group session_parser_02
> 2017-06-04 04:30:38,229 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-12_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-3,

[jira] [Commented] (KAFKA-3408) consumer rebalance fail

2017-08-07 Thread Ewen Cheslack-Postava (JIRA) 

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

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

[~airbots] Not sure if it is still an issue since it's quite an old JIRA. You 
could try to reproduce, but the bug is for the old Scala consumer that is now 
deprecated (as of 0.11.0.0). Bug fixes might still be accepted, but if you're 
looking to get into a newbie JIRA there might be some better options that 
address more recent issues. Let me know if you want some help finding one.

> consumer rebalance fail
> ---
>
> Key: KAFKA-3408
> URL: https://issues.apache.org/jira/browse/KAFKA-3408
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.9.0.1
> Environment: centos linux
>Reporter: zhongkai liu
>  Labels: newbie
>
> I use "/bin/kafka-console-consumer" command to start two consumers of group 
> "page_group",then the first conumer console report rebalance failure like 
> this:
> ERROR [page_view_group1_slave2-1458095694092-80c33086], error during 
> syncedRebalance (kafka.consumer.ZookeeperConsumerConnector)
> kafka.common.ConsumerRebalanceFailedException: 
> page_view_group1_slave2-1458095694092-80c33086 can't rebalance after 10 
> retries
> at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:660)
> at 
> kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anon$1.run(ZookeeperConsumerConnector.scala:579)



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

[jira] [Assigned] (KAFKA-5708) Update Jackson dependencies (from 2.8.5 to 2.9.x)

2017-08-07 Thread Ewen Cheslack-Postava (JIRA) 

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

Ewen Cheslack-Postava reassigned KAFKA-5708:


Assignee: Dejan Stojadinović  (was: Ewen Cheslack-Postava)

> Update Jackson dependencies (from 2.8.5 to 2.9.x)
> -
>
> Key: KAFKA-5708
> URL: https://issues.apache.org/jira/browse/KAFKA-5708
> Project: Kafka
>  Issue Type: Task
>  Components: build
>Reporter: Dejan Stojadinović
>Assignee: Dejan Stojadinović
>Priority: Blocker
> Fix For: 1.0.0
>
>
> In addition to update: remove deprecated version forcing for 
> 'jackson-annotations'
> *_Notes:_*
> * wait until [Jackson 
> 2.9.1|https://github.com/FasterXML/jackson/wiki/Jackson-Release-2.9.1] is 
> released (expected in September 2017)
> * inspired by pull request: https://github.com/apache/kafka/pull/3631



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

[jira] [Assigned] (KAFKA-5708) Update Jackson dependencies (from 2.8.5 to 2.9.x)

2017-08-07 Thread Ewen Cheslack-Postava (JIRA) 

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

Ewen Cheslack-Postava reassigned KAFKA-5708:


Assignee: Ewen Cheslack-Postava

> Update Jackson dependencies (from 2.8.5 to 2.9.x)
> -
>
> Key: KAFKA-5708
> URL: https://issues.apache.org/jira/browse/KAFKA-5708
> Project: Kafka
>  Issue Type: Task
>  Components: build
>Reporter: Dejan Stojadinović
>Assignee: Ewen Cheslack-Postava
>Priority: Blocker
> Fix For: 1.0.0
>
>
> In addition to update: remove deprecated version forcing for 
> 'jackson-annotations'
> *_Notes:_*
> * wait until [Jackson 
> 2.9.1|https://github.com/FasterXML/jackson/wiki/Jackson-Release-2.9.1] is 
> released (expected in September 2017)
> * inspired by pull request: https://github.com/apache/kafka/pull/3631



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

[jira] [Commented] (KAFKA-5701) Unit test shouldTogglePrepareForBulkLoadDuringRestoreCalls fails

2017-08-07 Thread ASF GitHub Bot (JIRA) 

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

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

Github user asfgit closed the pull request at:

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


> Unit test shouldTogglePrepareForBulkLoadDuringRestoreCalls fails
> 
>
> Key: KAFKA-5701
> URL: https://issues.apache.org/jira/browse/KAFKA-5701
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.11.0.1
>Reporter: Eno Thereska
>Assignee: Bill Bejeck
> Fix For: 1.0.0
>
>
> shouldTogglePrepareForBulkLoadDuringRestoreCalls occassionally fails. This 
> can be reproduced locally in IntelliJ when running in a loop. It fails two 
> ways:
> Way 1:
> --
> Exception in thread "Thread-6" java.lang.AssertionError: Condition not met 
> within timeout 1000. Did not prepare for bulk load
>   at org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:275)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStoreTest.assertRocksDBTurnsOnBulkLoading(RocksDBStoreTest.java:228)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStoreTest.access$000(RocksDBStoreTest.java:51)
>   at 
> org.apache.kafka.streams.state.internals.RocksDBStoreTest$1.run(RocksDBStoreTest.java:138)
>   at java.lang.Thread.run(Thread.java:745)
> Way 2:
> 
> org.apache.kafka.streams.state.internals.RocksDBStoreTest.shouldTogglePrepareForBulkLoadDuringRestoreCalls
> java.lang.AssertionError
> at org.junit.Assert.fail(Assert.java:86)
> at org.junit.Assert.assertTrue(Assert.java:41)
> at org.junit.Assert.assertTrue(Assert.java:52)
> at 
> org.apache.kafka.streams.state.internals.RocksDBStoreTest.shouldTogglePrepareForBulkLoadDuringRestoreCalls(RocksDBStoreTest.java:152)



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

[jira] [Commented] (KAFKA-5681) jarAll does not build all scala versions anymore.

2017-08-07 Thread ASF GitHub Bot (JIRA) 

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

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

Github user becketqin closed the pull request at:

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


> jarAll does not build all scala versions anymore.
> -
>
> Key: KAFKA-5681
> URL: https://issues.apache.org/jira/browse/KAFKA-5681
> Project: Kafka
>  Issue Type: Bug
>  Components: build
>Affects Versions: 0.11.0.0
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.11.0.1
>
>
> ./gradlew jarAll no longer builds jars for all scala versions. We should use 
> {{availableScalaVersions}} instead of {{defaultScalaVersions}} when build. We 
> probably should consider backporting the fix to 0.11.0.0.



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

[jira] [Commented] (KAFKA-5705) Kafka Server start failed and reports "unsafe memory access operation"

2017-08-07 Thread Waleed Fateem (JIRA) 

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

Waleed Fateem commented on KAFKA-5705:
--

[~hachikuji]

Just wondering if this is actually a different scenario that leads up to the 
same memory access issue. The error stack reported here looks a bit different 
than what was reported in KAFKA-5628.

> Kafka Server start failed and reports "unsafe memory access operation"
> --
>
> Key: KAFKA-5705
> URL: https://issues.apache.org/jira/browse/KAFKA-5705
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.10.2.0
>Reporter: Chen He
>
> [2017-08-02 15:50:23,361] FATAL Fatal error during KafkaServerStartable 
> startup. Prepare to shutdown (kafka.server.KafkaServerStartable)
> java.lang.InternalError: a fault occurred in a recent unsafe memory access 
> operation in compiled Java code
> at 
> kafka.log.TimeIndex$$anonfun$maybeAppend$1.apply$mcV$sp(TimeIndex.scala:128)
> at 
> kafka.log.TimeIndex$$anonfun$maybeAppend$1.apply(TimeIndex.scala:107)
> at 
> kafka.log.TimeIndex$$anonfun$maybeAppend$1.apply(TimeIndex.scala:107)
> at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:213)
> at kafka.log.TimeIndex.maybeAppend(TimeIndex.scala:107)
> at kafka.log.LogSegment.recover(LogSegment.scala:252)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:231)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:188)
> at 
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
> at 
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
> at 
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
> at kafka.log.Log.loadSegments(Log.scala:188)
> at kafka.log.Log.(Log.scala:116)
> at 
> kafka.log.LogManager$$anonfun$loadLogs$2$$anonfun$3$$anonfun$apply$10$$anonfun$apply$1.apply$mcV$sp(LogManager.scala:157)
> at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:57)
> 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)



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

[jira] [Commented] (KAFKA-5696) SourceConnector does not commit offset on rebalance

2017-08-07 Thread Oleg Kuznetsov (JIRA) 

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

Oleg Kuznetsov commented on KAFKA-5696:
---

Yes, you are right. 

Is there a chance that exactly-once delivery becomes possible since kafka 
supports this mode now?

> SourceConnector does not commit offset on rebalance
> ---
>
> Key: KAFKA-5696
> URL: https://issues.apache.org/jira/browse/KAFKA-5696
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Oleg Kuznetsov
>  Labels: newbie
> Fix For: 0.10.0.2
>
>
> I'm running SourceConnector, that reads files from storage and put data in 
> kafka. I want, in case of reconfiguration, offsets to be flushed. 
> Say, a file is completely processed, but source records are not yet committed 
> and in case of reconfiguration their offsets might be missing in store.
> Is it possible to force committing offsets on reconfiguration?



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

[jira] [Commented] (KAFKA-5676) MockStreamsMetrics should be in o.a.k.test

2017-08-07 Thread Guozhang Wang (JIRA) 

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

Guozhang Wang commented on KAFKA-5676:
--

Could you list the test cases that do depend on the {{Metrics}} object? If they 
do depend on the metrics then perhaps they should not use a mocked 
StreamsMetrics.

> MockStreamsMetrics should be in o.a.k.test
> --
>
> Key: KAFKA-5676
> URL: https://issues.apache.org/jira/browse/KAFKA-5676
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Chanchal Singh
>  Labels: newbie
>
> {{MockStreamsMetrics}}'s package should be `o.a.k.test` not 
> `o.a.k.streams.processor.internals`. 
> In addition, it should not require a {{Metrics}} parameter in its constructor 
> as it is only needed for its extended base class; the right way of mocking 
> should be implementing {{StreamsMetrics}} with mock behavior than extended a 
> real implementaion of {{StreamsMetricsImpl}}.



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

[jira] [Created] (KAFKA-5709) Improve logging to include errors from state-change log in server log

2017-08-07 Thread Alla Tumarkin (JIRA) 
Alla Tumarkin created KAFKA-5709:


 Summary: Improve logging to include errors from state-change log 
in server log
 Key: KAFKA-5709
 URL: https://issues.apache.org/jira/browse/KAFKA-5709
 Project: Kafka
  Issue Type: Improvement
  Components: log
Affects Versions: 0.10.2.0
Reporter: Alla Tumarkin


Problem
The following message was generated over and over again when running 
kafka-console-producer or kafka-console-consumer with SSL and ACLs enabled
{code}
WARN Error while fetching metadata with correlation id 1 : 
{test=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)` endlessly 
when I run kafka-console-producer or kafka-console-consumer
{code}
however server log (or authorizer log) did not indicate any problem.

Background
1) Initially, security.inter.broker.protocol setting was missing in 
server.properties, so connection was falling back to using plaintext port 
(9092), and only state-change.log actually contained the underlying error:
{code}
[2017-08-04 13:40:48,536] TRACE Controller 0 epoch 6 received response 
{error_code=31,partitions=[{topic=test,partition=0,error_code=31},{topic=__confluent.support.metrics,partition=0,error_code=31}]}
 for a request sent to broker localhost:9092 (id: 0 rack: null) 
(state.change.logger)
{code}
as per
https://kafka.apache.org/protocol#protocol_error_codes
{code}
CLUSTER_AUTHORIZATION_FAILED31  False   Cluster authorization failed.
{code}

2) After setting "security.inter.broker.protocol=SSL" the port changed to 
secure (9093) yet the error in state-change log did not go away:
{code}
[2017-08-04 13:49:38,462] TRACE Controller 0 epoch 7 received response 
{error_code=31} for a request sent to broker localhost:9093 (id: 0 rack: null) 
(state.change.logger)
{code}
and LEADER_NOT_AVAILABLE was still generated.

This time though, kafka-authorizer.log had a better indication of the problem:
{code}
[2017-08-04 18:17:46,770] DEBUG No acl found for resource 
Cluster:kafka-cluster, authorized = false (kafka.authorizer.logger)
[2017-08-04 18:17:46,770] DEBUG Principal = 
User:CN=Unknown,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown is Denied 
Operation = ClusterAction from host = 127.0.0.1 on resource = 
Cluster:kafka-cluster (kafka.authorizer.logger)
{code}
The issue being that topic metadata is not propagated successfully from 
controller to broker since the broker user doesn't have ClusterAction 
permission.
Fixed by
{code}
bin/kafka-acls --authorizer-properties zookeeper.connect=localhost:2181 --add 
--allow-principal 
"User:CN=Unknown,OU=Unknown,O=Unknown,L=Unknown,ST=Unknown,C=Unknown" 
--operation ClusterAction --cluster
{code}

Request
The debugging is tricky since the controller to broker logging is done in 
controller/state-change log, not in the main server log.
We need to improve the logging on this.



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

[jira] [Updated] (KAFKA-5704) Auto topic creation causes failure with older clusters

2017-08-07 Thread Randall Hauch (JIRA) 

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

Randall Hauch updated KAFKA-5704:
-
Fix Version/s: 0.11.0.1

> Auto topic creation causes failure with older clusters
> --
>
> Key: KAFKA-5704
> URL: https://issues.apache.org/jira/browse/KAFKA-5704
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.11.0.0
>Reporter: Ewen Cheslack-Postava
>Assignee: Randall Hauch
> Fix For: 0.11.0.1
>
>
> The new automatic internal topic creation always tries to check the topic and 
> create it if missing. However, older brokers that we should still be 
> compatible with don't support some requests that are used. This results in an 
> UnsupportedVersionException which some of the TopicAdmin code notes that it 
> can throw but then isn't caught in the initializers, causing the entire 
> process to fail.
> We should probably just catch it, log a message, and allow things to proceed 
> hoping that the user has already created the topics correctly (as we used to 
> do).



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

[jira] [Commented] (KAFKA-5704) Auto topic creation causes failure with older clusters

2017-08-07 Thread ASF GitHub Bot (JIRA) 

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

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

GitHub user rhauch opened a pull request:

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

KAFKA-5704 Corrected Connect distributed startup behavior to allow older 
brokers to auto-create topics

When a Connect distributed worker starts up talking with broker versions 
0.10.1.0 and later, it will use the AdminClient to look for the internal topics 
and attempt to create them if they are missing. Although the AdminClient was 
added in 0.11.0.0, the AdminClient uses APIs to create topics that existed in 
0.10.1.0 and later. This feature works as expected when Connect uses a broker 
version 0.10.1.0 or later.

However, when a Connect distributed worker starts up using a broker older 
than 0.10.1.0, the AdminClient is not able to find the required APIs and thus 
will throw an UnsupportedVersionException. Unfortunately, this exception is not 
caught and instead causes the Connect worker to fail even when the topics 
already exist.

This change handles the UnsupportedVersionException by logging a debug 
message and doing nothing. The existing producer logic will get information 
about the topics, which will cause the broker to create them if they don’t 
exist and broker auto-creation of topics is enabled. This is the same behavior 
that existed prior to 0.11.0.0, and so this change restores that behavior for 
brokers older than 0.10.1.0.

This change also adds a system test that verifies Connect works with a 
variety of brokers and is able to run source and sink connectors. The test 
verifies that Connect can read from the internal topics when the connectors are 
restarted.

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

$ git pull https://github.com/rhauch/kafka kafka-5704

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

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


commit 0d45fd113eeaff3844742181191db2cc508353fd
Author: Randall Hauch 
Date:   2017-08-07T19:32:29Z

KAFKA-5704 Corrected Connect distributed startup behavior to allow older 
brokers to auto-create topics

When a Connect distributed worker starts up talking with broker versions 
0.10.1.0 and later, it will use the AdminClient to look for the internal topics 
and attempt to create them if they are missing. Although the AdminClient was 
added in 0.11.0.0, the AdminClient uses APIs to create topics that existed in 
0.10.1.0 and later. This feature works as expected when Connect uses a broker 
version 0.10.1.0 or later.

However, when a Connect distributed worker starts up using a broker older 
than 0.10.1.0, the AdminClient is not able to find the required APIs and thus 
will throw an UnsupportedVersionException. Unfortunately, this exception is not 
caught and instead causes the Connect worker to fail even when the topics 
already exist.

This change handles the UnsupportedVersionException by logging a debug 
message and doing nothing. The existing producer logic will get information 
about the topics, which will cause the broker to create them if they don’t 
exist and broker auto-creation of topics is enabled. This is the same behavior 
that existed prior to 0.11.0.0, and so this change restores that behavior for 
brokers older than 0.10.1.0.

This change also adds a system test that verifies Connect works with a 
variety of brokers and is able to run source and sink connectors. The test 
verifies that Connect can read from the internal topics when the connectors are 
restarted.




> Auto topic creation causes failure with older clusters
> --
>
> Key: KAFKA-5704
> URL: https://issues.apache.org/jira/browse/KAFKA-5704
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.11.0.0
>Reporter: Ewen Cheslack-Postava
>Assignee: Randall Hauch
>
> The new automatic internal topic creation always tries to check the topic and 
> create it if missing. However, older brokers that we should still be 
> compatible with don't support some requests that are used. This results in an 
> UnsupportedVersionException which some of the TopicAdmin code notes that it 
> can throw but then isn't caught in the initializers, causing the entire 
> process to fail.
> We should probably just catch it, log a message, and allow things to proceed 
> hoping that the user has already created the topics correctly (as we used to 
> do).



--
This message was sent by Atlassian JIRA

[jira] [Commented] (KAFKA-5696) SourceConnector does not commit offset on rebalance

2017-08-07 Thread Randall Hauch (JIRA) 

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

Randall Hauch commented on KAFKA-5696:
--

Look inside the {{while}} loop within that {{WorkerSourceTask.execute()}} 
method. That while loop calls {{poll()}}, and when the task stops the while 
loop breaks and the finally block you quote is called. So, no, offsets are 
committed periodically and not after every {{poll()}}.

However, it does seem that this finally block should be called when the task 
stops, even when part of the rebalance. How have you verified that offsets are 
not flushed? One thing is to turn on debug logging, as the {{WorkerSourceTask}} 
logs quite a few debugging messages during the offset.

> SourceConnector does not commit offset on rebalance
> ---
>
> Key: KAFKA-5696
> URL: https://issues.apache.org/jira/browse/KAFKA-5696
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Oleg Kuznetsov
>  Labels: newbie
> Fix For: 0.10.0.2
>
>
> I'm running SourceConnector, that reads files from storage and put data in 
> kafka. I want, in case of reconfiguration, offsets to be flushed. 
> Say, a file is completely processed, but source records are not yet committed 
> and in case of reconfiguration their offsets might be missing in store.
> Is it possible to force committing offsets on reconfiguration?



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

[jira] [Commented] (KAFKA-5696) SourceConnector does not commit offset on rebalance

2017-08-07 Thread Oleg Kuznetsov (JIRA) 

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

Oleg Kuznetsov commented on KAFKA-5696:
---

[~rhauch] Thank you for detailed explanation. 

But aren't offsets committed also at the end of every poll() method? 
I found this in WorkerSourceTask.execute() finally block.

{code:java}
 } finally {
// It should still be safe to commit offsets since any exception 
would have
// simply resulted in not getting more records but all the existing 
records should be ok to flush
// and commit offsets. Worst case, task.flush() will also throw an 
exception causing the offset commit
// to fail.
commitOffsets();
}
{code}

> SourceConnector does not commit offset on rebalance
> ---
>
> Key: KAFKA-5696
> URL: https://issues.apache.org/jira/browse/KAFKA-5696
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Oleg Kuznetsov
>  Labels: newbie
> Fix For: 0.10.0.2
>
>
> I'm running SourceConnector, that reads files from storage and put data in 
> kafka. I want, in case of reconfiguration, offsets to be flushed. 
> Say, a file is completely processed, but source records are not yet committed 
> and in case of reconfiguration their offsets might be missing in store.
> Is it possible to force committing offsets on reconfiguration?



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

[jira] [Resolved] (KAFKA-5705) Kafka Server start failed and reports "unsafe memory access operation"

2017-08-07 Thread Jason Gustafson (JIRA) 

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

Jason Gustafson resolved KAFKA-5705.

Resolution: Duplicate

Marking this as a duplicate of KAFKA-5628 since the exception trace looks 
identical. Please reopen if you disagree.

> Kafka Server start failed and reports "unsafe memory access operation"
> --
>
> Key: KAFKA-5705
> URL: https://issues.apache.org/jira/browse/KAFKA-5705
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Affects Versions: 0.10.2.0
>Reporter: Chen He
>
> [2017-08-02 15:50:23,361] FATAL Fatal error during KafkaServerStartable 
> startup. Prepare to shutdown (kafka.server.KafkaServerStartable)
> java.lang.InternalError: a fault occurred in a recent unsafe memory access 
> operation in compiled Java code
> at 
> kafka.log.TimeIndex$$anonfun$maybeAppend$1.apply$mcV$sp(TimeIndex.scala:128)
> at 
> kafka.log.TimeIndex$$anonfun$maybeAppend$1.apply(TimeIndex.scala:107)
> at 
> kafka.log.TimeIndex$$anonfun$maybeAppend$1.apply(TimeIndex.scala:107)
> at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:213)
> at kafka.log.TimeIndex.maybeAppend(TimeIndex.scala:107)
> at kafka.log.LogSegment.recover(LogSegment.scala:252)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:231)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:188)
> at 
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
> at 
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
> at 
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
> at kafka.log.Log.loadSegments(Log.scala:188)
> at kafka.log.Log.(Log.scala:116)
> at 
> kafka.log.LogManager$$anonfun$loadLogs$2$$anonfun$3$$anonfun$apply$10$$anonfun$apply$1.apply$mcV$sp(LogManager.scala:157)
> at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:57)
> 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)



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

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

2017-08-07 Thread Jason Gustafson (JIRA) 

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

Jason Gustafson commented on KAFKA-5212:


[~evis] The fetcher and the heartbeat share the same network client, so when 
we're blocked in {{poll}} inside {{Fetcher}}, the heartbeat thread can't send 
any heartbeats. In retrospect, we probably should have given the heartbeat 
thread its own network client, but sharing made its initial implementation a 
bit simpler. For now, maybe we just need to ensure that 
{{sendListOffsetRequest}} takes the heartbeat thread into account when setting 
its poll timeout.

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



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

[jira] [Assigned] (KAFKA-5581) Streams can be smarter in deciding when to create changelog topics for state stores

2017-08-07 Thread Mariam John (JIRA) 

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

Mariam John reassigned KAFKA-5581:
--

Assignee: Mariam John

> Streams can be smarter in deciding when to create changelog topics for state 
> stores
> ---
>
> Key: KAFKA-5581
> URL: https://issues.apache.org/jira/browse/KAFKA-5581
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Mariam John
>  Labels: architecture, performance
>
> Today Streams make all state stores to be backed by a changelog topic by 
> default unless users overrides it by {{disableLogging}} when creating the 
> state store / materializing the KTable. However there are a few cases where a 
> separate changelog topic would not be required as we can re-use an existing 
> topic for that. A few examples:
> There are a few places where the materialized store do not need a separate 
> changelog topic, for example:
> 1) If a KTable is read directly from a source topic, and is materialized i.e. 
> {code}
> table1 = builder.table("topic1", "store1")`.
> {code}
> In this case {{table1}}'s changelog topic can just be {{topic1}}, and we do 
> not need to create a separate {{table1-changelog}} topic.
> 2) if a KTable is materialized and then sent directly into a sink topic with 
> the same key, e.g.
> {code}
> table1 = stream.groupBy(...).aggregate("state1").to("topic2");
> {code}
> In this case {{state1}}'s changelog topic can just be {{topic2}}, and we do 
> not need to create a separate {{state1-changelog}} topic anymore;
> 3) if a KStream is materialized for joins where the streams are directly from 
> a topic, e.g.:
> {code}
> stream1 = builder.stream("topic1");
> stream2 = builder.stream("topic2");
> stream3 = stream1.join(stream2, windows);  // stream1 and stream2 are 
> materialized with a changelog topic
> {code}
> Since stream materialization is append-only we do not need a changelog for 
> the state store as well but can just use the source {{topic1}} and {{topic2}}.
> 4) When you have some simple transformation operations or even join 
> operations that generated new KTables, and which needs to be materialized 
> with a state store, you can use the changelog topic of the previous KTable 
> and applies the transformation logic upon restoration instead of creating a 
> new changelog topic. For example:
> {code}
> table1 = builder.table("topic1");
> table2 = table1.filter(..).join(table3); // table2 needs to be materialized 
> for joining
> {code}
> We can set the {{getter}} function of table2's materialized store, say 
> {{state2}} to be reading from {{topic1}} and then apply the filter operator, 
> instead of creating a new {{state2-changelog}} topic in this case.
> 5) more use cases ...
> We can come up with a general internal impl optimizations to determine when / 
> how to set the changelog topic for those materialized stores at the runtime 
> startup when generating the topology.



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

[jira] [Assigned] (KAFKA-5697) StreamThread.close() need to interrupt the stream threads to break the loop

2017-08-07 Thread Mariam John (JIRA) 

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

Mariam John reassigned KAFKA-5697:
--

Assignee: Mariam John

> StreamThread.close() need to interrupt the stream threads to break the loop
> ---
>
> Key: KAFKA-5697
> URL: https://issues.apache.org/jira/browse/KAFKA-5697
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Mariam John
>
> In {{StreamThread.close()}} we currently do nothing but set the state, hoping 
> the stream thread may eventually check it and shutdown itself. However, under 
> certain scenarios the thread may get blocked within a single loop and hence 
> will never check on this state enum. For example, it's {{consumer.poll}} call 
> trigger {{ensureCoordinatorReady()}} which will block until the coordinator 
> can be found. If the coordinator broker is never up and running then the 
> Stream instance will be blocked forever.
> A simple way to produce this issue is to start the work count demo without 
> starting the ZK / Kafka broker, and then it will get stuck in a single loop 
> and even `ctrl-C` will not stop it since its set state will never be read by 
> the thread:
> {code}
> [2017-08-03 15:17:39,981] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> [2017-08-03 15:17:40,046] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> [2017-08-03 15:17:40,101] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> [2017-08-03 15:17:40,206] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> [2017-08-03 15:17:40,261] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> [2017-08-03 15:17:40,366] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> [2017-08-03 15:17:40,472] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> ^C[2017-08-03 15:17:40,580] WARN Connection to node -1 could not be 
> established. Broker may not be available. 
> (org.apache.kafka.clients.NetworkClient)
> {code}



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

[jira] [Commented] (KAFKA-5503) Idempotent producer ignores shutdown while fetching ProducerId

2017-08-07 Thread Jason Gustafson (JIRA) 

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

Jason Gustafson commented on KAFKA-5503:


[~evis] Apologies for the late response. The issue is that 
{{NetworkClientUtils.sendAndReceive}} and {{Sender.maybeWaitForProducerId}} 
prevents the sender thread shutting down. Even if you wakeup the selector, the 
loop will just throw the sender right back into poll. At a minimum, we should 
be checking the {{running}} flag in the loop inside {{maybeWaitForProducerId}}.

> Idempotent producer ignores shutdown while fetching ProducerId
> --
>
> Key: KAFKA-5503
> URL: https://issues.apache.org/jira/browse/KAFKA-5503
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Affects Versions: 0.11.0.0
>Reporter: Jason Gustafson
>Assignee: Evgeny Veretennikov
> Fix For: 0.11.0.1
>
>
> When using the idempotent producer, we initially block the sender thread 
> while we attempt to get the ProducerId. During this time, a concurrent call 
> to close() will be ignored.



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

[jira] [Commented] (KAFKA-5430) new consumers getting data for revoked partitions

2017-08-07 Thread Jason Gustafson (JIRA) 

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

Jason Gustafson commented on KAFKA-5430:


[~liorchaga] Thanks for the report. I think this problem may have been fixed in 
KAFKA-5154 (the commit itself is a bit more descriptive: 
https://github.com/apache/kafka/commit/1b16acaaa181ceb214d84e70b8ddc146af9c0c5c).
 Is there any chance you could try again with the 0.11.0.0 client?

> new consumers getting data for revoked partitions
> -
>
> Key: KAFKA-5430
> URL: https://issues.apache.org/jira/browse/KAFKA-5430
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, consumer
>Affects Versions: 0.10.2.0
>Reporter: Lior Chaga
> Attachments: consumer-thread.log, consumer-thread.log, 
> kafka_trace.log.gz
>
>
> Due to bad configuration applied to network components, we experienced issues 
> with communication between kafka brokers (causing under replication) as well 
> as producers/consumers not being able to work against kafka.
> The symptoms on the consumer were many errors of the following form:
> {code}
> 2017-06-04 04:27:35,200 ERROR [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] TaboolaKafkaConsumer [] - Failed 
> committing to kafka topicPartitions 
> [requestlegacy-2,requestlegacy-0,requestlegacy-1] 
> org.apache.kafka.clients.consumer.RetriableCommitFailedException: Offset 
> commit failed with a retriable exception. You should retry committing offsets.
> Caused by: org.apache.kafka.common.errors.DisconnectException
> {code}
> So far so good. However, upon network recovery, there were several rebalance 
> operations, which eventually resulted in only one consumer (#14) being 
> assigned with all topic partitions (at this case we're talking about a 
> consumer groups for which all consumers are running in same process):
> {code}
> 2017-06-04 04:27:02,168 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-14_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-8, requestlegacy-9] 
> for group session_parser_02
> 2017-06-04 04:27:04,208 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-15_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-10, requestlegacy-11] 
> for group session_parser_02
> 2017-06-04 04:27:18,167 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-12_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-3, requestlegacy-4, 
> requestlegacy-5] for group session_parser_02
> 2017-06-04 04:27:20,232 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-2, requestlegacy-0, 
> requestlegacy-1] for group session_parser_02
> 2017-06-04 04:27:20,236 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-15_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-9, requestlegacy-10, 
> requestlegacy-11] for group session_parser_02
> 2017-06-04 04:27:20,237 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-12_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-3, requestlegacy-4, requestlegacy-5] 
> for group session_parser_02
> 2017-06-04 04:27:20,237 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-14_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-6, requestlegacy-7, requestlegacy-8] 
> for group session_parser_02
> 2017-06-04 04:27:20,332 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] ConsumerCoordinator [] - Setting 
> newly assigned partitions [requestlegacy-2, requestlegacy-0, requestlegacy-1] 
> for group session_parser_02
> 2017-06-04 04:28:52,368 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-13_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-6, requestlegacy-7] 
> for group session_parser_02
> 2017-06-04 04:29:15,201 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-11_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-2, requestlegacy-0, 
> requestlegacy-1] for group session_parser_02
> 2017-06-04 04:30:22,379 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-14_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-6, requestlegacy-7, 
> requestlegacy-8] for group session_parser_02
> 2017-06-04 04:30:24,431 INFO  [Kafka Topics Cosumer 
> requestlegacy.consumer-15_session_parser_02] ConsumerCoordinator [] - 
> Revoking previously assigned partitions [requestlegacy-9, requestlegacy-10, 
>

[jira] [Commented] (KAFKA-2526) Console Producer / Consumer's serde config is not working

2017-08-07 Thread Ewen Cheslack-Postava (JIRA) 

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

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

[~ardahal] Yeah, I think the scope of the bug is just unclear. Adding better 
errors would be a useful short-term stopgap solution (I think you just get a 
cast exception or something like that if you try to set these to something 
besides ByteArraySerializer, but I haven't double checked). The longer term 
solution with a KIP is obviously a lot more involved, but you should feel free 
to work on it; it would be a useful improvement. Since there would be two 
separate steps, you'd probably want to either file a separate JIRA for the 
better error messages and leave this one to the KIP or just file that fix as a 
MINOR PR.

> Console Producer / Consumer's serde config is not working
> -
>
> Key: KAFKA-2526
> URL: https://issues.apache.org/jira/browse/KAFKA-2526
> Project: Kafka
>  Issue Type: Bug
>Reporter: Guozhang Wang
>Assignee: Mayuresh Gharat
>  Labels: newbie
>
> Although in the console producer one can specify the key value serializer, 
> they are actually not used since 1) it always serialize the input string as 
> String.getBytes (hence always pre-assume the string serializer) and 2) it is 
> actually only passed into the old producer. The same issues exist in console 
> consumer.
> In addition the configs in the console producer is messy: we have 1) some 
> config values exposed as cmd parameters, and 2) some config values in 
> --producer-property and 3) some in --property.
> It will be great to clean the configs up in both console producer and 
> consumer, and put them into a single --property parameter which could 
> possibly take a file to reading in property values as well, and only leave 
> --new-producer as the other command line parameter.



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

[jira] [Commented] (KAFKA-5516) Formatting verifiable producer/consumer output in a similar fashion

2017-08-07 Thread ASF GitHub Bot (JIRA) 

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

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

Github user asfgit closed the pull request at:

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


> Formatting verifiable producer/consumer output in a similar fashion
> ---
>
> Key: KAFKA-5516
> URL: https://issues.apache.org/jira/browse/KAFKA-5516
> Project: Kafka
>  Issue Type: Improvement
>  Components: tools
>Reporter: Paolo Patierno
>Assignee: Paolo Patierno
>Priority: Trivial
> Fix For: 1.0.0
>
>
> Hi,
> following the proposal to have verifiable producer/consumer providing a very 
> similar output where the "timestamp" is always the first column followed by 
> "name" event and then all the specific data for such event.
> It includes a verifiable producer refactoring for having that in the same way 
> as verifiable consumer.
> Thanks,
> Paolo



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

[jira] [Commented] (KAFKA-4879) KafkaConsumer.position may hang forever when deleting a topic

2017-08-07 Thread ASF GitHub Bot (JIRA) 

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

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

GitHub user baluchicken opened a pull request:

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

KAFKA-4879 KafkaConsumer.position may hang forever when deleting a topic

@guozhangwang  I created this PR but it lacks the test. I also has the test 
but I wanted to ask which Test file should be the best candidate for this?
Also there is a TODO line which I created because that method already has a 
timeout parameter, what do you think how can we proceed there?

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

$ git pull https://github.com/baluchicken/kafka-1 KAFKA-4879

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

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


commit 211b2118b0ec5fd7ff5b499ef17c92b0d31e76e3
Author: Balint Molnar 
Date:   2017-08-07T14:54:34Z

KAFKA-4879 KafkaConsumer.position may hang forever when deleting a topic




> KafkaConsumer.position may hang forever when deleting a topic
> -
>
> Key: KAFKA-4879
> URL: https://issues.apache.org/jira/browse/KAFKA-4879
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.10.2.0
>Reporter: Shixiong Zhu
>Assignee: Balint Molnar
> Fix For: 1.0.0
>
>
> KafkaConsumer.position may hang forever when deleting a topic. The problem is 
> this line 
> https://github.com/apache/kafka/blob/022bf129518e33e165f9ceefc4ab9e622952d3bd/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L374
> The timeout is "Long.MAX_VALUE", and it will just retry forever for 
> UnknownTopicOrPartitionException.
> Here is a reproducer
> {code}
> import org.apache.kafka.clients.consumer.KafkaConsumer;
> import org.apache.kafka.common.TopicPartition;
> import org.apache.kafka.common.serialization.StringDeserializer;
> import java.util.Collections;
> import java.util.Properties;
> import java.util.Set;
> public class KafkaReproducer {
>   public static void main(String[] args) {
> // Make sure "delete.topic.enable" is set to true.
> // Please create the topic test with "3" partitions manually.
> // The issue is gone when there is only one partition.
> String topic = "test";
> Properties props = new Properties();
> props.put("bootstrap.servers", "localhost:9092");
> props.put("group.id", "testgroup");
> props.put("value.deserializer", StringDeserializer.class.getName());
> props.put("key.deserializer", StringDeserializer.class.getName());
> props.put("enable.auto.commit", "false");
> KafkaConsumer kc = new KafkaConsumer(props);
> kc.subscribe(Collections.singletonList(topic));
> kc.poll(0);
> Set partitions = kc.assignment();
> System.out.println("partitions: " + partitions);
> kc.pause(partitions);
> kc.seekToEnd(partitions);
> System.out.println("please delete the topic in 30 seconds");
> try {
>   // Sleep 30 seconds to give us enough time to delete the topic.
>   Thread.sleep(3);
> } catch (InterruptedException e) {
>   e.printStackTrace();
> }
> System.out.println("sleep end");
> for (TopicPartition p : partitions) {
>   System.out.println(p + " offset: " + kc.position(p));
> }
> System.out.println("cannot reach here");
> kc.close();
>   }
> }
> {code}



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

[jira] [Comment Edited] (KAFKA-5696) SourceConnector does not commit offset on rebalance

2017-08-07 Thread Randall Hauch (JIRA) 

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

Randall Hauch edited comment on KAFKA-5696 at 8/7/17 2:29 PM:
--

Kafka Connect currently guarantees *at least once*, for a couple of reasons.

First, Kafka Connect currently commits offsets *periodically* (configurable via 
worker's {{offset.flush.interval.ms}} property that defaults to 60 seconds, and 
Kafka Connect will always restart from the last *committed offset*. The most 
recent offsets are always committed just prior to graceful shutdown (and should 
be just prior to a rebalance, per this issue), but any unexpected failure will 
likely mean that the last the connector has produced records that were written 
to Kafka but not yet reflected in committed offsets. 

Second, when the Kafka Connect worker writes source records to Kafka, the 
broker may successfully accept and write a batch of source records to the 
specified number of workers and send an acknowledgement. However, if a network 
glitch prevents the Kafka Connect worker's producer from receiving the broker's 
acknowledgement, the producer will resend the batch of source records. 

Bottom line, there is always a chance of the connector producing records that 
are written to Kafka but not reflected in committed offsets. 


was (Author: rhauch):
Kafka Connect currently guarantees *at least once*, for a couple of reasons.

First, Kafka Connect currently commits offsets *periodically* (configurable via 
worker's {{offset.flush.interval.ms}} property that defaults to 60 seconds, and 
Kafka Connect will always restart from the last *committed offset*. The most 
recent offsets are always committed just prior to graceful shutdown (and should 
be just prior to a rebalance, per this issue), but any unexpected failure will 
likely mean that the last the connector has produced records that were written 
to Kafka but not reflected in committed offsets. 

Second, when the Kafka Connect worker writes source records to Kafka, the 
broker may successfully accept and write a batch of source records to the 
specified number of workers and send an acknowledgement. However, if a network 
glitch prevents the Kafka Connect worker's producer from receiving the broker's 
acknowledgement, the producer will resend the batch of source records. 

Bottom line, there is always a chance of the connector producing records that 
are written to Kafka but not reflected in committed offsets. 

> SourceConnector does not commit offset on rebalance
> ---
>
> Key: KAFKA-5696
> URL: https://issues.apache.org/jira/browse/KAFKA-5696
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Oleg Kuznetsov
>  Labels: newbie
> Fix For: 0.10.0.2
>
>
> I'm running SourceConnector, that reads files from storage and put data in 
> kafka. I want, in case of reconfiguration, offsets to be flushed. 
> Say, a file is completely processed, but source records are not yet committed 
> and in case of reconfiguration their offsets might be missing in store.
> Is it possible to force committing offsets on reconfiguration?



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

[jira] [Updated] (KAFKA-5696) SourceConnector does not commit offset on rebalance

2017-08-07 Thread Randall Hauch (JIRA) 

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

Randall Hauch updated KAFKA-5696:
-
Summary: SourceConnector does not commit offset on rebalance  (was: 
SourceConnector does not commit offset on reconfiguration)

> SourceConnector does not commit offset on rebalance
> ---
>
> Key: KAFKA-5696
> URL: https://issues.apache.org/jira/browse/KAFKA-5696
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Oleg Kuznetsov
>  Labels: newbie
> Fix For: 0.10.0.2
>
>
> I'm running SourceConnector, that reads files from storage and put data in 
> kafka. I want, in case of reconfiguration, offsets to be flushed. 
> Say, a file is completely processed, but source records are not yet committed 
> and in case of reconfiguration their offsets might be missing in store.
> Is it possible to force committing offsets on reconfiguration?



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

[jira] [Commented] (KAFKA-5696) SourceConnector does not commit offset on reconfiguration

2017-08-07 Thread Randall Hauch (JIRA) 

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

Randall Hauch commented on KAFKA-5696:
--

Kafka Connect currently guarantees *at least once*, for a couple of reasons.

First, Kafka Connect currently commits offsets *periodically* (configurable via 
worker's {{offset.flush.interval.ms}} property that defaults to 60 seconds, and 
Kafka Connect will always restart from the last *committed offset*. The most 
recent offsets are always committed just prior to graceful shutdown (and should 
be just prior to a rebalance, per this issue), but any unexpected failure will 
likely mean that the last the connector has produced records that were written 
to Kafka but not reflected in committed offsets. 

Second, when the Kafka Connect worker writes source records to Kafka, the 
broker may successfully accept and write a batch of source records to the 
specified number of workers and send an acknowledgement. However, if a network 
glitch prevents the Kafka Connect worker's producer from receiving the broker's 
acknowledgement, the producer will resend the batch of source records. 

Bottom line, there is always a chance of the connector producing records that 
are written to Kafka but not reflected in committed offsets. 

> SourceConnector does not commit offset on reconfiguration
> -
>
> Key: KAFKA-5696
> URL: https://issues.apache.org/jira/browse/KAFKA-5696
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Oleg Kuznetsov
>  Labels: newbie
> Fix For: 0.10.0.2
>
>
> I'm running SourceConnector, that reads files from storage and put data in 
> kafka. I want, in case of reconfiguration, offsets to be flushed. 
> Say, a file is completely processed, but source records are not yet committed 
> and in case of reconfiguration their offsets might be missing in store.
> Is it possible to force committing offsets on reconfiguration?



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

[jira] [Commented] (KAFKA-4879) KafkaConsumer.position may hang forever when deleting a topic

2017-08-07 Thread Balint Molnar (JIRA) 

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

Balint Molnar commented on KAFKA-4879:
--

[~guozhang] I debugged this issue, my findings are the following:
* The partition size for the topic is need to be bigger than num.partitions 
configuration value.
* This problem only occurs when delete.topic.enable is set to true.
* This problem only occurs when auto.create.topics.enable set to true. (if this 
one is set to false then it does not matter how many partitions are set in 
num.partitions the position will hang indefinitely.)
* The main problem here is: during the consumer.position call the consumer will 
recreate the topic with the num.partitions partitions and if it is smaller than 
the original topic's partition num it will hang indefinitely. 

> KafkaConsumer.position may hang forever when deleting a topic
> -
>
> Key: KAFKA-4879
> URL: https://issues.apache.org/jira/browse/KAFKA-4879
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.10.2.0
>Reporter: Shixiong Zhu
>Assignee: Balint Molnar
> Fix For: 1.0.0
>
>
> KafkaConsumer.position may hang forever when deleting a topic. The problem is 
> this line 
> https://github.com/apache/kafka/blob/022bf129518e33e165f9ceefc4ab9e622952d3bd/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L374
> The timeout is "Long.MAX_VALUE", and it will just retry forever for 
> UnknownTopicOrPartitionException.
> Here is a reproducer
> {code}
> import org.apache.kafka.clients.consumer.KafkaConsumer;
> import org.apache.kafka.common.TopicPartition;
> import org.apache.kafka.common.serialization.StringDeserializer;
> import java.util.Collections;
> import java.util.Properties;
> import java.util.Set;
> public class KafkaReproducer {
>   public static void main(String[] args) {
> // Make sure "delete.topic.enable" is set to true.
> // Please create the topic test with "3" partitions manually.
> // The issue is gone when there is only one partition.
> String topic = "test";
> Properties props = new Properties();
> props.put("bootstrap.servers", "localhost:9092");
> props.put("group.id", "testgroup");
> props.put("value.deserializer", StringDeserializer.class.getName());
> props.put("key.deserializer", StringDeserializer.class.getName());
> props.put("enable.auto.commit", "false");
> KafkaConsumer kc = new KafkaConsumer(props);
> kc.subscribe(Collections.singletonList(topic));
> kc.poll(0);
> Set partitions = kc.assignment();
> System.out.println("partitions: " + partitions);
> kc.pause(partitions);
> kc.seekToEnd(partitions);
> System.out.println("please delete the topic in 30 seconds");
> try {
>   // Sleep 30 seconds to give us enough time to delete the topic.
>   Thread.sleep(3);
> } catch (InterruptedException e) {
>   e.printStackTrace();
> }
> System.out.println("sleep end");
> for (TopicPartition p : partitions) {
>   System.out.println(p + " offset: " + kc.position(p));
> }
> System.out.println("cannot reach here");
> kc.close();
>   }
> }
> {code}



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

[jira] [Commented] (KAFKA-5684) KStreamPrintProcessor as customized KStreamPeekProcessor

2017-08-07 Thread ASF GitHub Bot (JIRA) 

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

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

GitHub user ppatierno opened a pull request:

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

KAFKA-5684: KStreamPrintProcessor as customized KStreamPeekProcessor

This PR is intended for having KStreamPrint derived from KStreamPeek and 
avoiding the "instance of" check on byte[ ] every process call.

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

$ git pull https://github.com/ppatierno/kafka kafka-5684

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

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


commit a516a08bf22cdd53feb4ada50c73dc5a08715b74
Author: Paolo Patierno 
Date:   2017-08-07T12:55:16Z

Refactored KStreamPrint as derived from KStreamPeek
Removed the "instance of" check for byte[] in every KStreamPrint process, 
it's up to a default mapper now
Updated KStreamPrint tests adapting to the new internal structure for 
KStreamPrint and PrintForeachAction




> KStreamPrintProcessor as customized KStreamPeekProcessor
> 
>
> Key: KAFKA-5684
> URL: https://issues.apache.org/jira/browse/KAFKA-5684
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Paolo Patierno
>Assignee: Paolo Patierno
>Priority: Minor
>
> Hi,
> the {{KStreamPrintProcessor}} is implemented from scratch (from the 
> {{AbstractProcessor}}) and the same for the related supplier.
> It looks to me that it's just a special {{KStreamPeekProcessor}} with 
> forwardDownStream to false and that allows the possibility to specify Serdes 
> instances used if key/values are bytes.
> At same time used by a {{print()}} method it provides a fast way to print 
> data flowing through the pipeline (while using just {{peek()}} you need to 
> write the code).
> I think that it could be useful to refactoring the {{KStreamPrintProcessor}} 
> as derived from the {{KStreamPeekProcessor}} customizing its behavior.
> Thanks,
> Paolo.



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

[jira] [Commented] (KAFKA-5138) MirrorMaker doesn't exit on send failure occasionally

2017-08-07 Thread Dustin Cote (JIRA) 

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

Dustin Cote commented on KAFKA-5138:


[~tamas.mate] I have no plans to work on this at this time. It's all yours :)

> MirrorMaker doesn't exit on send failure occasionally
> -
>
> Key: KAFKA-5138
> URL: https://issues.apache.org/jira/browse/KAFKA-5138
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.10.2.0
>Reporter: Dustin Cote
>  Labels: newbie
>
> MirrorMaker with abort.on.send.failure=true does not always exit if the 
> producer closes. Here is the logic that happens:
> First we encounter a problem producing and force the producer to close
> {code}
> [2017-04-10 07:17:25,137] ERROR Error when sending message to topic 
> mytopicwith key: 20 bytes, value: 314 bytes with error: 
> (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
> org.apache.kafka.common.errors.TimeoutException: Expiring 47 record(s) for 
> mytopic-2: 30879 ms has passed since last append
> [2017-04-10 07:17:25,170] INFO Closing producer due to send failure. 
> (kafka.tools.MirrorMaker$)
> [2017-04-10 07:17:25,170] INFO Closing the Kafka producer with timeoutMillis 
> = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] INFO Proceeding to force close the producer since 
> pending requests could not be completed within timeout 0 ms. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] DEBUG The Kafka producer has closed. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] ERROR Error when sending message to topic mytopic 
> with key: 20 bytes, value: 313 bytes with error: 
> (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
> org.apache.kafka.common.errors.TimeoutException: Expiring 47 record(s) for 
> mytopic-2: 30879 ms has passed since last append
> [2017-04-10 07:17:25,170] INFO Closing producer due to send failure. 
> (kafka.tools.MirrorMaker$)
> [2017-04-10 07:17:25,170] INFO Closing the Kafka producer with timeoutMillis 
> = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] INFO Proceeding to force close the producer since 
> pending requests could not be completed within timeout 0 ms. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] DEBUG The Kafka producer has closed. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> {code}
> All good there. Then we can't seem to close the producer nicely after about 
> 15 seconds and so it is forcefully killed:
> {code}
> [2017-04-10 07:17:39,778] ERROR Error when sending message to topic 
> mytopic.subscriptions with key: 70 bytes, value: null with error: 
> (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
> java.lang.IllegalStateException: Producer is closed forcefully.
> at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:522)
> at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:502)
> at 
> org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:147)
> at java.lang.Thread.run(Unknown Source)
> [2017-04-10 07:17:39,778] INFO Closing producer due to send failure. 
> (kafka.tools.MirrorMaker$)
> [2017-04-10 07:17:39,778] INFO Closing the Kafka producer with timeoutMillis 
> = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:39,778] INFO Proceeding to force close the producer since 
> pending requests could not be completed within timeout 0 ms. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:39,778] DEBUG The Kafka producer has closed. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:39,779] DEBUG Removed sensor with name connections-closed: 
> (org.apache.kafka.common.metrics.Metrics)
> {code}
> After removing some metric sensors for awhile this happens:
> {code}
> [2017-04-10 07:17:39,780] DEBUG Removed sensor with name node-3.latency 
> (org.apache.kafka.common.metrics.Metrics)
> [2017-04-10 07:17:39,780] DEBUG Shutdown of Kafka producer I/O thread has 
> completed. (org.apache.kafka.clients.producer.internals.Sender)
> [2017-04-10 07:17:41,852] DEBUG Sending Heartbeat request for group 
> mirror-maker-1491619052-teab1-1 to coordinator myhost1:9092 (id: 2147483643 
> rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-04-10 07:17:41,953] DEBUG Received successful Heartbeat response for 
> group mirror-maker-1491619052-teab1-1 
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-04-10 07:17:44,875] DEBUG Sending Heartbeat request for group 
>

[jira] [Commented] (KAFKA-5138) MirrorMaker doesn't exit on send failure occasionally

2017-08-07 Thread Tamas Mate (JIRA) 

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

Tamas Mate commented on KAFKA-5138:
---

Hi [~cote],
have you planned to work on this? Because I would like to check this out.

> MirrorMaker doesn't exit on send failure occasionally
> -
>
> Key: KAFKA-5138
> URL: https://issues.apache.org/jira/browse/KAFKA-5138
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.10.2.0
>Reporter: Dustin Cote
>  Labels: newbie
>
> MirrorMaker with abort.on.send.failure=true does not always exit if the 
> producer closes. Here is the logic that happens:
> First we encounter a problem producing and force the producer to close
> {code}
> [2017-04-10 07:17:25,137] ERROR Error when sending message to topic 
> mytopicwith key: 20 bytes, value: 314 bytes with error: 
> (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
> org.apache.kafka.common.errors.TimeoutException: Expiring 47 record(s) for 
> mytopic-2: 30879 ms has passed since last append
> [2017-04-10 07:17:25,170] INFO Closing producer due to send failure. 
> (kafka.tools.MirrorMaker$)
> [2017-04-10 07:17:25,170] INFO Closing the Kafka producer with timeoutMillis 
> = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] INFO Proceeding to force close the producer since 
> pending requests could not be completed within timeout 0 ms. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] DEBUG The Kafka producer has closed. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] ERROR Error when sending message to topic mytopic 
> with key: 20 bytes, value: 313 bytes with error: 
> (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
> org.apache.kafka.common.errors.TimeoutException: Expiring 47 record(s) for 
> mytopic-2: 30879 ms has passed since last append
> [2017-04-10 07:17:25,170] INFO Closing producer due to send failure. 
> (kafka.tools.MirrorMaker$)
> [2017-04-10 07:17:25,170] INFO Closing the Kafka producer with timeoutMillis 
> = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] INFO Proceeding to force close the producer since 
> pending requests could not be completed within timeout 0 ms. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:25,170] DEBUG The Kafka producer has closed. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> {code}
> All good there. Then we can't seem to close the producer nicely after about 
> 15 seconds and so it is forcefully killed:
> {code}
> [2017-04-10 07:17:39,778] ERROR Error when sending message to topic 
> mytopic.subscriptions with key: 70 bytes, value: null with error: 
> (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
> java.lang.IllegalStateException: Producer is closed forcefully.
> at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.abortBatches(RecordAccumulator.java:522)
> at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.abortIncompleteBatches(RecordAccumulator.java:502)
> at 
> org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:147)
> at java.lang.Thread.run(Unknown Source)
> [2017-04-10 07:17:39,778] INFO Closing producer due to send failure. 
> (kafka.tools.MirrorMaker$)
> [2017-04-10 07:17:39,778] INFO Closing the Kafka producer with timeoutMillis 
> = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:39,778] INFO Proceeding to force close the producer since 
> pending requests could not be completed within timeout 0 ms. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:39,778] DEBUG The Kafka producer has closed. 
> (org.apache.kafka.clients.producer.KafkaProducer)
> [2017-04-10 07:17:39,779] DEBUG Removed sensor with name connections-closed: 
> (org.apache.kafka.common.metrics.Metrics)
> {code}
> After removing some metric sensors for awhile this happens:
> {code}
> [2017-04-10 07:17:39,780] DEBUG Removed sensor with name node-3.latency 
> (org.apache.kafka.common.metrics.Metrics)
> [2017-04-10 07:17:39,780] DEBUG Shutdown of Kafka producer I/O thread has 
> completed. (org.apache.kafka.clients.producer.internals.Sender)
> [2017-04-10 07:17:41,852] DEBUG Sending Heartbeat request for group 
> mirror-maker-1491619052-teab1-1 to coordinator myhost1:9092 (id: 2147483643 
> rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-04-10 07:17:41,953] DEBUG Received successful Heartbeat response for 
> group mirror-maker-1491619052-teab1-1 
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-04-10 07:17:44,875] DEBUG Sending Heartbeat request for group 
>

[jira] [Commented] (KAFKA-5617) Update to the list of third-party clients

2017-08-07 Thread Daniel Schierbeck (JIRA) 

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

Daniel Schierbeck commented on KAFKA-5617:
--



On Fri, Aug 4, 2017 at 6:54 PM Jason Gustafson (JIRA) 



> Update to the list of third-party clients
> -
>
> Key: KAFKA-5617
> URL: https://issues.apache.org/jira/browse/KAFKA-5617
> Project: Kafka
>  Issue Type: Wish
>Reporter: Daniel Schierbeck
>
> I'd like to have the list of Ruby client libraries updated to reflect the 
> current state of affairs.
> * ruby-kafka is no longer compatible with Kafka 0.8, but is compatible with 
> 0.9+. It should probably be moved to the top, since it's the only actively 
> maintained low-level library – all other libraries are either unmaintained or 
> are opinionated frameworks built on top of ruby-kafka.
> * I'd like to add Racecar (https://github.com/zendesk/racecar), a simple 
> opinionated framework built on top of ruby-kafka. It's an extraction from a 
> production Zendesk code base, and so is already pretty battle tested.
> I'm the author of both ruby-kafka and Racecar.



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

[jira] [Resolved] (KAFKA-5707) Remove useless `--force` option for both TopicCommand and ConfigCommand

2017-08-07 Thread huxihx (JIRA) 

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

huxihx resolved KAFKA-5707.
---
Resolution: Not A Bug

For the sake of compatibility, just keep `--force` in both classes. Closed this 
jira then.

> Remove useless `--force` option for both TopicCommand and ConfigCommand
> ---
>
> Key: KAFKA-5707
> URL: https://issues.apache.org/jira/browse/KAFKA-5707
> Project: Kafka
>  Issue Type: Bug
>  Components: admin
>Affects Versions: 0.11.0.0
>Reporter: huxihx
>Assignee: huxihx
>Priority: Minor
>
> `TopicCommand` and `ConfigCommand` do expose an option named `--force` which 
> suppresses console prompts, but both classes do not actually use it. Should 
> remove it from the usage description.



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

[jira] [Commented] (KAFKA-5707) Remove useless `--force` option for both TopicCommand and ConfigCommand

2017-08-07 Thread ASF GitHub Bot (JIRA) 

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

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

Github user huxihx closed the pull request at:

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


> Remove useless `--force` option for both TopicCommand and ConfigCommand
> ---
>
> Key: KAFKA-5707
> URL: https://issues.apache.org/jira/browse/KAFKA-5707
> Project: Kafka
>  Issue Type: Bug
>  Components: admin
>Affects Versions: 0.11.0.0
>Reporter: huxihx
>Assignee: huxihx
>Priority: Minor
>
> `TopicCommand` and `ConfigCommand` do expose an option named `--force` which 
> suppresses console prompts, but both classes do not actually use it. Should 
> remove it from the usage description.



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