Build failed in Jenkins: kafka-trunk-jdk7 #2025

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-4848: Fix retryWithBackoff deadlock issue

--
[...truncated 163.82 KB...]

kafka.producer.ProducerTest > testSendNullMessage STARTED

kafka.producer.ProducerTest > testSendNullMessage PASSED

kafka.producer.ProducerTest > testUpdateBrokerPartitionInfo STARTED

kafka.producer.ProducerTest > testUpdateBrokerPartitionInfo PASSED

kafka.producer.ProducerTest > testSendWithDeadBroker STARTED

kafka.producer.ProducerTest > testSendWithDeadBroker PASSED

kafka.tools.MirrorMakerIntegrationTest > testCommaSeparatedRegex STARTED

kafka.tools.MirrorMakerIntegrationTest > testCommaSeparatedRegex PASSED

kafka.tools.ReplicaVerificationToolTest > testReplicaBufferVerifyChecksum 
STARTED

kafka.tools.ReplicaVerificationToolTest > testReplicaBufferVerifyChecksum PASSED

kafka.tools.ConsoleConsumerTest > shouldLimitReadsToMaxMessageLimit STARTED

kafka.tools.ConsoleConsumerTest > shouldLimitReadsToMaxMessageLimit PASSED

kafka.tools.ConsoleConsumerTest > shouldParseValidNewConsumerValidConfig STARTED

kafka.tools.ConsoleConsumerTest > shouldParseValidNewConsumerValidConfig PASSED

kafka.tools.ConsoleConsumerTest > shouldStopWhenOutputCheckErrorFails STARTED

kafka.tools.ConsoleConsumerTest > shouldStopWhenOutputCheckErrorFails PASSED

kafka.tools.ConsoleConsumerTest > 
shouldParseValidNewSimpleConsumerValidConfigWithStringOffset STARTED

kafka.tools.ConsoleConsumerTest > 
shouldParseValidNewSimpleConsumerValidConfigWithStringOffset PASSED

kafka.tools.ConsoleConsumerTest > shouldParseConfigsFromFile STARTED

kafka.tools.ConsoleConsumerTest > shouldParseConfigsFromFile PASSED

kafka.tools.ConsoleConsumerTest > 
shouldParseValidNewSimpleConsumerValidConfigWithNumericOffset STARTED

kafka.tools.ConsoleConsumerTest > 
shouldParseValidNewSimpleConsumerValidConfigWithNumericOffset PASSED

kafka.tools.ConsoleConsumerTest > testDefaultConsumer STARTED

kafka.tools.ConsoleConsumerTest > testDefaultConsumer PASSED

kafka.tools.ConsoleConsumerTest > shouldParseValidOldConsumerValidConfig STARTED

kafka.tools.ConsoleConsumerTest > shouldParseValidOldConsumerValidConfig PASSED

kafka.tools.ConsoleProducerTest > testParseKeyProp STARTED

kafka.tools.ConsoleProducerTest > testParseKeyProp PASSED

kafka.tools.ConsoleProducerTest > testValidConfigsOldProducer STARTED

kafka.tools.ConsoleProducerTest > testValidConfigsOldProducer PASSED

kafka.tools.ConsoleProducerTest > testInvalidConfigs STARTED

kafka.tools.ConsoleProducerTest > testInvalidConfigs PASSED

kafka.tools.ConsoleProducerTest > testValidConfigsNewProducer STARTED

kafka.tools.ConsoleProducerTest > testValidConfigsNewProducer PASSED

kafka.tools.MirrorMakerTest > 
testDefaultMirrorMakerMessageHandlerWithNoTimestampInSourceMessage STARTED

kafka.tools.MirrorMakerTest > 
testDefaultMirrorMakerMessageHandlerWithNoTimestampInSourceMessage PASSED

kafka.tools.MirrorMakerTest > testDefaultMirrorMakerMessageHandler STARTED

kafka.tools.MirrorMakerTest > testDefaultMirrorMakerMessageHandler PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes PASSED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic STARTED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic PASSED

kafka.common.ConfigTest > testInvalidGroupIds STARTED

kafka.common.ConfigTest > testInvalidGroupIds PASSED

kafka.common.ConfigTest > testInvalidClientIds STARTED

kafka.common.ConfigTest > testInvalidClientIds PASSED

kafka.common.ZkNodeChangeNotificationListenerTest > testProcessNotification 
STARTED

kafka.common.ZkNodeChangeNotificationListenerTest > testProcessNotification 
PASSED

kafka.common.TopicTest > testInvalidTopicNames STARTED

kafka.common.TopicTest > testInvalidTopicNames PASSED

kafka.common.TopicTest > testTopicHasCollision STARTED

kafka.co

[jira] [Resolved] (KAFKA-4848) Stream thread getting into deadlock state while trying to get rocksdb lock in retryWithBackoff

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-4848.
--
   Resolution: Fixed
Fix Version/s: 0.11.0.0

> Stream thread getting into deadlock state while trying to get rocksdb lock in 
> retryWithBackoff
> --
>
> Key: KAFKA-4848
> URL: https://issues.apache.org/jira/browse/KAFKA-4848
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Sachin Mittal
>Assignee: Sachin Mittal
> Fix For: 0.11.0.0
>
> Attachments: thr-1
>
>
> We see a deadlock state when streams thread to process a task takes longer 
> than MAX_POLL_INTERVAL_MS_CONFIG time. In this case this threads partitions 
> are assigned to some other thread including rocksdb lock. When it tries to 
> process the next task it cannot get rocks db lock and simply keeps waiting 
> for that lock forever.
> in retryWithBackoff for AbstractTaskCreator we have a backoffTimeMs = 50L.
> If it does not get lock the we simply increase the time by 10x and keep 
> trying inside the while true loop.
> We need to have a upper bound for this backoffTimeM. If the time is greater 
> than  MAX_POLL_INTERVAL_MS_CONFIG and it still hasn't got the lock means this 
> thread's partitions are moved somewhere else and it may not get the lock 
> again.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4848) Stream thread getting into deadlock state while trying to get rocksdb lock in retryWithBackoff

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Stream thread getting into deadlock state while trying to get rocksdb lock in 
> retryWithBackoff
> --
>
> Key: KAFKA-4848
> URL: https://issues.apache.org/jira/browse/KAFKA-4848
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Sachin Mittal
>Assignee: Sachin Mittal
> Attachments: thr-1
>
>
> We see a deadlock state when streams thread to process a task takes longer 
> than MAX_POLL_INTERVAL_MS_CONFIG time. In this case this threads partitions 
> are assigned to some other thread including rocksdb lock. When it tries to 
> process the next task it cannot get rocks db lock and simply keeps waiting 
> for that lock forever.
> in retryWithBackoff for AbstractTaskCreator we have a backoffTimeMs = 50L.
> If it does not get lock the we simply increase the time by 10x and keep 
> trying inside the while true loop.
> We need to have a upper bound for this backoffTimeM. If the time is greater 
> than  MAX_POLL_INTERVAL_MS_CONFIG and it still hasn't got the lock means this 
> thread's partitions are moved somewhere else and it may not get the lock 
> again.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2642: KAFKA-4848: Fix retryWithBackoff deadlock issue

[asfgit]

Github user asfgit closed the pull request at:

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


---
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-4750) KeyValueIterator returns null values

[Kamal Chandraprakash (JIRA)]

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

Kamal Chandraprakash reassigned KAFKA-4750:
---

Assignee: Kamal Chandraprakash

> KeyValueIterator returns null values
> 
>
> Key: KAFKA-4750
> URL: https://issues.apache.org/jira/browse/KAFKA-4750
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.1.1
>Reporter: Michal Borowiecki
>Assignee: Kamal Chandraprakash
>  Labels: newbie
>
> The API for ReadOnlyKeyValueStore.range method promises the returned iterator 
> will not return null values. However, after upgrading from 0.10.0.0 to 
> 0.10.1.1 we found null values are returned causing NPEs on our side.
> I found this happens after removing entries from the store and I found 
> resemblance to SAMZA-94 defect. The problem seems to be as it was there, when 
> deleting entries and having a serializer that does not return null when null 
> is passed in, the state store doesn't actually delete that key/value pair but 
> the iterator will return null value for that key.
> When I modified our serilizer to return null when null is passed in, the 
> problem went away. However, I believe this should be fixed in kafka streams, 
> perhaps with a similar approach as SAMZA-94.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

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

[Apache Jenkins Server]

See 


Changes:

[junrao] KAFKA-4894; Fix findbugs "default character set in use" warnings

--
[...truncated 1.46 MB...]
org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
shouldPerformAllQueriesWithCachingDisabled PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
shouldPerformRangeQueriesWithCachingDisabled STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
shouldPerformRangeQueriesWithCachingDisabled PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
shouldCloseOpenIteratorsWhenStoreClosedAndThrowInvalidStateStoreOnHasNextAndNext
 STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
shouldCloseOpenIteratorsWhenStoreClosedAndThrowInvalidStateStoreOnHasNextAndNext
 PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > testSize 
STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > testSize 
PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutIfAbsent STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutIfAbsent PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testRestoreWithDefaultSerdes STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testRestoreWithDefaultSerdes PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > testRestore 
STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > testRestore 
PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRange STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRange PASSED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRangeWithDefaultSerdes STARTED

org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreTest > 
testPutGetRangeWithDefaultSerdes PASSED

org.apache.kafka.streams.KeyValueTest > shouldHaveSaneEqualsAndHashCode STARTED

org.apache.kafka.streams.KeyValueTest > shouldHaveSaneEqualsAndHashCode PASSED

org.apache.kafka.streams.KafkaStreamsTest > testIllegalMetricsConfig STARTED

org.apache.kafka.streams.KafkaStreamsTest > testIllegalMetricsConfig PASSED

org.apache.kafka.streams.KafkaStreamsTest > shouldNotGetAllTasksWhenNotRunning 
STARTED

org.apache.kafka.streams.KafkaStreamsTest > shouldNotGetAllTasksWhenNotRunning 
PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
testInitializesAndDestroysMetricsReporters STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
testInitializesAndDestroysMetricsReporters PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndPartitionerWhenNotRunning STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndPartitionerWhenNotRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > testLegalMetricsConfig STARTED

org.apache.kafka.streams.KafkaStreamsTest > testLegalMetricsConfig PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndSerializerWhenNotRunning STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetTaskWithKeyAndSerializerWhenNotRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldReturnFalseOnCloseWhenThreadsHaventTerminated STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldReturnFalseOnCloseWhenThreadsHaventTerminated PASSED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetAllTasksWithStoreWhenNotRunning STARTED

org.apache.kafka.streams.KafkaStreamsTest > 
shouldNotGetAllTasksWithStoreWhenNotRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartOnceClosed STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartOnceClosed PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCleanup STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCleanup PASSED

org.apache.kafka.streams.KafkaStreamsTest > testNumberDefaultMetrics STARTED

org.apache.kafka.streams.KafkaStreamsTest > testNumberDefaultMetrics PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCloseIsIdempotent STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCloseIsIdempotent PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCannotCleanupWhileRunning 
STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCannotCleanupWhileRunning PASSED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartTwice STARTED

org.apache.kafka.streams.KafkaStreamsTest > testCannotStartTwice PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableJoin STARTED

org.apache.kafka.

[jira] [Commented] (KAFKA-2729) Cached zkVersion not equal to that in zookeeper, broker not recovering.

[Ronghua Lin (JIRA)]

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

Ronghua Lin commented on KAFKA-2729:


[~junrao], we also have this problem in a small cluster which has 3 brokers, 
running Kafka 0.10.1.1. When it happened, the logs of each broker look like 
this:
{code:title=broker 2 | borderStyle=solid}
[2017-03-20 01:03:48,903] INFO [Group Metadata Manager on Broker 2]: Removed 0 
expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2017-03-20 01:13:27,283] INFO Creating /controller (is it secure? false) 
(kafka.utils.ZKCheckedEphemeral)
[2017-03-20 01:13:27,293] INFO Result of znode creation is: OK 
(kafka.utils.ZKCheckedEphemeral)
[2017-03-20 01:13:27,294] INFO 2 successfully elected as leader 
(kafka.server.ZookeeperLeaderElector)
[2017-03-20 01:13:28,203] INFO re-registering broker info in ZK for broker 2 
(kafka.server.KafkaHealthcheck$SessionExpireListener)
[2017-03-20 01:13:28,205] INFO Creating /brokers/ids/2 (is it secure? false) 
(kafka.utils.ZKCheckedEphemeral)
[2017-03-20 01:13:28,218] INFO Result of znode creation is: OK 
(kafka.utils.ZKCheckedEphemeral)
[2017-03-20 01:13:28,219] INFO Registered broker 2 at path /brokers/ids/2 with 
addresses: PLAINTEXT -> EndPoint(x, ,PLAINTEXT) (kafka.utils.ZkUtils)
[2017-03-20 01:13:28,219] INFO done re-registering broker 
(kafka.server.KafkaHealthcheck$SessionExpireListener)
[2017-03-20 01:13:28,220] INFO Subscribing to /brokers/topics path to watch for 
new topics (kafka.server.KafkaHealthcheck$SessionExpireListener)
[2017-03-20 01:13:28,224] INFO New leader is 2 
(kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
[2017-03-20 01:13:28,227] INFO New leader is 2 
(kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
[2017-03-20 01:13:38,812] INFO Partition [topic1,1] on broker 2: Shrinking ISR 
for partition [topic1,1] from 0,2,1 to 2,1 (kafka.cluster.Partition)
[2017-03-20 01:13:38,825] INFO Partition [topic1,1] on broker 2: Cached 
zkVersion [6] not equal to that in zookeeper, skip updating ISR 
(kafka.cluster.Partition)
[2017-03-20 01:13:38,825] INFO Partition [topic2,1] on broker 2: Shrinking ISR 
for partition [topic2,1] from 0,2,1 to 2,1 (kafka.cluster.Partition)
[2017-03-20 01:13:38,835] INFO Partition [topic2,1] on broker 2: Cached 
zkVersion [6] not equal to that in zookeeper, skip updating ISR 
(kafka.cluster.Partition)
[2017-03-20 01:13:38,835] INFO Partition [topic3,0] on broker 2: Shrinking ISR 
for partition [topic3,0] from 0,2,1 to 2,1 (kafka.cluster.Partition)
[2017-03-20 01:13:38,847] INFO Partition [topic3,0] on broker 2: Cached 
zkVersion [6] not equal to that in zookeeper, skip updating ISR 
(kafka.cluster.Partition)

{code}

{code:title=broker 1 | borderStyle=solid}
[2017-03-20 01:03:38,255] INFO [Group Metadata Manager on Broker 1]: Removed 0 
expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2017-03-20 01:13:27,451] INFO New leader is 2 
(kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
[2017-03-20 01:13:27,490] INFO re-registering broker info in ZK for broker 1 
(kafka.server.KafkaHealthcheck$SessionExpireListener)
[2017-03-20 01:13:27,491] INFO Creating /brokers/ids/1 (is it secure? false) 
(kafka.utils.ZKCheckedEphemeral)
[2017-03-20 01:13:27,503] INFO Result of znode creation is: OK 
(kafka.utils.ZKCheckedEphemeral)
[2017-03-20 01:13:27,503] INFO Registered broker 1 at path /brokers/ids/1 with 
addresses: PLAINTEXT -> EndPoint(,,PLAINTEXT) (kafka.utils.ZkUtils)
[2017-03-20 01:13:27,504] INFO done re-registering broker 
(kafka.server.KafkaHealthcheck$SessionExpireListener)
[2017-03-20 01:13:27,504] INFO Subscribing to /brokers/topics path to watch for 
new topics (kafka.server.KafkaHealthcheck$SessionExpireListener)
[2017-03-20 01:13:27,508] INFO New leader is 2 
(kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
[2017-03-20 01:13:38,134] INFO Partition [__consumer_offsets,40] on broker 1: 
Shrinking ISR for partition [__consumer_offsets,40] from 1,0 to 1 
(kafka.cluster.Partition)
[2017-03-20 01:13:38,155] INFO Partition [__consumer_offsets,40] on broker 1: 
Cached zkVersion [2] not equal to that in zookeeper, skip updating ISR 
(kafka.cluster.Partition)
[2017-03-20 01:13:38,156] INFO Partition [__consumer_offsets,0] on broker 1: 
Shrinking ISR for partition [__consumer_offsets,0] from 1,0 to 1 
(kafka.cluster.Partition)
[2017-03-20 01:13:38,161] INFO Partition [__consumer_offsets,0] on broker 1: 
Cached zkVersion [2] not equal to that in zookeeper, skip updating ISR 
(kafka.cluster.Partition)
[2017-03-20 01:13:38,162] INFO Partition [__consumer_offsets,12] on broker 1: 
Shrinking ISR for partition [__consumer_offsets,12] from 1,0 to 1 
(kafka.cluster.Partition)
[2017-03-20 01:13:38,170] INFO Partition [__consumer_offsets,12] on broker 1: 
Cached zkVersion [2] not equal to that in zook

[jira] [Commented] (KAFKA-4921) AssignedPartition should implement equals

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax commented on KAFKA-4921:


I am not sure if we need this. {{StreamPartitionAssignor}} is never 
hashed/compared or similar. It would add dead code if I am not wrong. Or can 
you point out where equals or hashcode would be called?

> AssignedPartition should implement equals
> -
>
> Key: KAFKA-4921
> URL: https://issues.apache.org/jira/browse/KAFKA-4921
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Marc Juchli
>Priority: Minor
>
> We ran FundBug, which resulted in the "Bad practice warning":
>   
> Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
> In class 
> org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition
> In method 
> org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition.compareTo(StreamPartitionAssignor$AssignedPartition)
> At StreamPartitionAssignor.java:[line 75]



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Build failed in Jenkins: kafka-trunk-jdk7 #2024

[Apache Jenkins Server]

See 


Changes:

[junrao] KAFKA-4894; Fix findbugs "default character set in use" warnings

--
[...truncated 856.72 KB...]
org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldThrowStreamsExceptionNoResetSpecified PASSED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldOnlyReadRecordsWhereEarliestSpecified STARTED

org.apache.kafka.streams.integration.KStreamsFineGrainedAutoResetIntegrationTest
 > shouldOnlyReadRecordsWhereEarliestSpecified PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduceSessionWindows STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduceSessionWindows PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduce STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduce PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregate STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregate PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCount STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCount PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldGroupByKey STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldGroupByKey PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduceWindowed STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduceWindowed PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountSessionWindows STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountSessionWindows PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregateWindowed STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregateWindowed PASSED

org.apache.kafka.streams.integration.FanoutIntegrationTest > 
shouldFanoutTheInput[0] STARTED

org.apache.kafka.streams.integration.FanoutIntegrationTest > 
shouldFanoutTheInput[0] PASSED

org.apache.kafka.streams.integration.FanoutIntegrationTest > 
shouldFanoutTheInput[1] STARTED

org.apache.kafka.streams.integration.FanoutIntegrationTest > 
shouldFanoutTheInput[1] PASSED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testShouldReadFromRegexAndNamedTopics STARTED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testShouldReadFromRegexAndNamedTopics PASSED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testRegexMatchesTopicsAWhenCreated STARTED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testRegexMatchesTopicsAWhenCreated PASSED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testMultipleConsumersCanReadFromPartitionedTopic STARTED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testMultipleConsumersCanReadFromPartitionedTopic PASSED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testRegexMatchesTopicsAWhenDeleted STARTED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testRegexMatchesTopicsAWhenDeleted PASSED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testNoMessagesSentExceptionFromOverlappingPatterns STARTED

org.apache.kafka.streams.integration.RegexSourceIntegrationTest > 
testNoMessagesSentExceptionFromOverlappingPatterns PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin PASSED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableIntegrationTest > 
shouldKStreamGlobalKTableJoin PASSED

org.apache.kafka.streams.integration.InternalTopicIntegrationTest > 
shouldCompactTopicsForStateChangelogs STARTED

org.apache.kafka.streams.integration.InternalTopicIntegrationTest > 
shouldCompactTopicsForStateChangelogs PASSED

org.apache.kafka.streams.integration.InternalTopicIntegrationTest > 
shouldUseCompactAndDeleteForWindowStoreChangelogs STARTED

org.apache.kafka.streams.integration.InternalTopicIntegrationTest > 
shouldUseCompactAndDeleteForWindowStoreChangelogs PASSED

org.apache.kafka.streams.integration.KStreamRepartitionJoinTest > 
shouldCorrectlyRepartitionOnJoinOperations[0] STARTED

org.apache.kafka.streams.integration.KStreamRepartitionJoinTest > 
shouldCorrectlyRepartitionOnJoinOperations[0] PASSED

Re: [DISCUSS] KIP-129: Kafka Streams Exactly-Once Semantics

[Eno Thereska]

Hi Matthias,

I'd like to see some more info on how you propose to handle transactions that 
involve state stores in the KIP itself. The design doc has info about various 
optimisations like RocksDb snapshots and transactions and such, but will there 
be a user-visible interface that indicates whether a store has snapshot and/or 
transactional capabilities? If a user plugs in another store, what guarantees 
are they expected to get? 

Will a V1 design that relies on plain store recovery from Kafka for each 
transaction abort be good enough, or usable? If your dataset is large (e.g., 
200GB) the recovery time might be so large as to effectively render that Kafka 
Streams instance unavailable for tens of minutes. You mention that is not a 
regression to what we currently have, however it seems to me we might have a 
regression of sorts: currently we pay the recovery price for a Kafka Streams 
instance failure. Now we might pay it for a transaction failure. Will 
transaction failures be more or less common than the previous types of 
failures? I'd like to see this addressed.

Thanks
Eno



> On 15 Mar 2017, at 22:09, Matthias J. Sax  wrote:
> 
> Just a quick follow up:
> 
> Our overall proposal is, to implement KIP-129 as is as a “Stream EoS
> 1.0” version. The raised concerns are all valid, but hard to quantify at
> the moment. Implementing KIP-129, that provides a clean design, allows
> us to gain more insight in the performance implications. This enables
> us, to make an educated decision, if the “producer per task” model
> perform wells or not, and if a switch to a “producer per thread” model
> is mandatory.
> 
> We also want to point out, that we can move incrementally from "producer
> per task" to "producer per thread" design or apply some incremental
> improvements to "producer per task" (as discussed in the doc). Thus,
> there is not issue with regard to upgrading.
> 
> 
> -Matthias
> 
> 
> 
> On 3/15/17 2:36 PM, Matthias J. Sax wrote:
>> Hi,
>> 
>> I want to pick up this thread again. As there are some concerns about
>> the "producer per task" design, we did write up an alternative "producer
>> per thread" design and discuss pros/cons of both approaches:
>> 
>> https://docs.google.com/document/d/1CfOJaa6mdg5o7pLf_zXISV4oE0ZeMZwT_sG1QWgL4EE
>> 
>> 
>> Looking forward to your feedback.
>> 
>> 
>> -Matthias
>> 
>> 
>> On 3/10/17 3:24 AM, Damian Guy wrote:
>>> Hi Matthias,
>>> 
>>> Thanks for the response. I agree with you regarding the use of
>>> PartitionGrouper to reduce the number of tasks. It would be good to have an
>>> idea of any additional load on the brokers as we increase the number of
>>> tasks and therefore producers.
>>> 
>>> Thanks,
>>> Damian
>>> 
>>> On Wed, 8 Mar 2017 at 01:45 Matthias J. Sax  wrote:
>>> 
 Damian, Jun,
 
 Thanks for your input.
 
 
 About Performance test:
 
 I can follow up with more performance tests using more partitions and
 also collecting broker metrics.
 
 However, I want to highlight again, that even if 1000+ partitions would
 be problematic, one can simply implement PartitionGrouper interface and
 reduce the number of tasks to 250 or 100... So I am not sure, if we
 should block this KIP, even if there might be some performance penalty
 for currently single partitioned tasks.
 
 About memory usage. JXM max-heap and max-off-heap did report 256MB and
 133MB for all experiments (thus I did not put it in the spreadsheet).
 Thus, using 100 producers (each using a max of 32MB of memory) was not
 an issue with regard to memory consumption. I did not track "current
 head/off-heap" memory as this would require a more advance test setup to
 monitor it over time. If you think this would be required, we can do
 some tests though.
 
 However, as 256 MB was enough memory, and there are other components
 next to the producers using memory, I don't expect a severely increased
 memory usage. Producer allocate memory on-demand, and if load is shared
 over multiple producers, overall memory usage should stay the same as a
 single producer should allocate less memory.
 
 
 About Batching:
 
 As you can see from the benchmarks (in the detailed view -- I also added
 some graphs to the summary now) the average batch size gets slightly
 decrease with an increased number of partitions. However, there is no
 big difference between "producer per thread" and "producer per task"
 scenario.
 
 
 About acks:
 
 This is covered by KIP-98 already. If idempotent producer is use, it's
 required to set max.in.flight.requests.per.connection=1 and retries > 0
 -- otherwise a config exception will be thrown. For transactions, it's
 further required that acks=-1 to avoid a config exception.
 
 Other bits, like min.isr, replication.factor, etc. (ie, all broker/topic
 configs) are out of scope, and it's user re

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

[Eno Thereska]

Hi Marc,

Could you add more information in the motivation of the KIP as to what problems 
this would solve? I can see how it can be done, but I don't yet grok why it's 
useful. The KIP should contain more pain points/problems and pose this as a 
solution. I know it's a small modification, but it's still important to have a 
good motivation IMO.

Thanks
Eno

> On 20 Mar 2017, at 18:25, Matthias J. Sax  wrote:
> 
> Sound reasonable Damian, but I guess, that's more a PR than KIP discussion.
> 
> @Marc, I guess you can start a VOTE thread if there is no further feedback.
> 
> 
> -Matthias
> 
> On 3/20/17 7:06 AM, Damian Guy wrote:
>> Hi Marc,
>> 
>> Thanks for the KIP. It mostly looks good to me. The only thing i'd change
>> is using a null argument to use a default mapping. IMO it would be better
>> if the existing print() method delegates to the new one supplying a
>> KeyValueMapper that does the right thing.
>> 
>> Thanks,
>> Damian
>> 
>> On Sat, 18 Mar 2017 at 14:25 Marc Juchli  wrote:
>> 
>>> Thanks!
>>> 
>>> I wanted to PING this thread. Not sure what the next steps of the KIP
>>> process are?
>>> 
>>> Kind regards,
>>> Marc
>>> 
>>> On Wed, Mar 15, 2017 at 9:13 PM Matthias J. Sax 
>>> wrote:
>>> 
 Thanks for updating the KIP.
 
 It's in very good shape IMHO and I support this idea!
 
 
 
 -Matthias
 
 
 On 3/15/17 3:05 AM, Marc Juchli wrote:
> Dear Matthias,
> 
> The KIP is updated. I think it now contains all the information on that
> page.
> 
> Marc
> 
> On Mon, Mar 13, 2017 at 9:37 PM Matthias J. Sax >>> 
> wrote:
> 
>> Marc,
>> 
>> Thanks for the KIP.
>> 
>> Can you please update the KIP in a way such that it is self contained.
>> Right now, you link to all kind of other places making it hard to read
>> the KIP.
>> 
>> The KIP should be the "center of truth" -- if there is important
>> information elsewhere, please c&p it into the KIP.
>> 
>> 
>> Thanks a lot!
>> 
>> 
>> -Matthias
>> 
>> 
>> 
>> On 3/13/17 1:30 PM, Matthias J. Sax wrote:
>>> Can you please add the KIP to this table:
>>> 
>>> 
>> 
 
>>> https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals#KafkaImprovementProposals-KIPsunderdiscussion
>>> 
>>> Thanks,
>>> 
>>> Matthias
>>> 
>>> 
>>> On 3/13/17 8:08 AM, Marc Juchli wrote:
 Dear all,
 
 The following describes KIP-132, which I just created. See:
 
>> 
 
>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-132+-+Augment+KStream.print+to+allow+extra+parameters+in+the+printed+string
 
 Motivation
 
 As for now, KStream#print leads to a predefined output where key and
>> value are
 printed with comma separation.
 KAFKA-4830 
 suggests
>> to
 extend print in a way that it takes KeyValueMapper as a parameter.
 This will allow a user to change outputs according to the users
 demand.
 Public Interfaces
 
 The affected interface is KStream, which needs to be extended with
>> another
 overloaded version of print:
 
 void print(final Serde keySerde,
   final Serde valSerde,
   final String streamName,
   final KeyValueMapper mapper);
 
 Proposed Changes
 
 See pull request GH-2669  as well as
>> KAFKA-4772
 .
 
 Compatibility, Deprecation, and Migration Plan
 
 The extension of print will not introduce compatibility issues – we
 can
 maintain the current output by keeping the current output format as
>>> a
 default (if mapper was not set):
 
 if(mapper == null) {
printStream.println("[" + streamName + "]: " + keyToPrint + " ,
>>> "
 + valueToPrint);
 } else {
printStream.println("[" + streamName + "]: " +
 mapper.apply(keyToPrint, valueToPrint));
 }
 
 
 
 Kind regards,
 Marc
 
>>> 
>> 
>> 
> 
 
 
>>> 
>> 
> 

[jira] [Commented] (KAFKA-4894) Fix findbugs "default character set in use" warnings

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Fix findbugs "default character set in use" warnings
> 
>
> Key: KAFKA-4894
> URL: https://issues.apache.org/jira/browse/KAFKA-4894
> Project: Kafka
>  Issue Type: Bug
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
> Fix For: 0.11.0.0
>
>
> Fix findbugs "default character set in use" warnings



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2683: KAFKA-4894. Fix findbugs "default character set in...

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Resolved] (KAFKA-4894) Fix findbugs "default character set in use" warnings

[Jun Rao (JIRA)]

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

Jun Rao resolved KAFKA-4894.

   Resolution: Fixed
Fix Version/s: 0.11.0.0

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

> Fix findbugs "default character set in use" warnings
> 
>
> Key: KAFKA-4894
> URL: https://issues.apache.org/jira/browse/KAFKA-4894
> Project: Kafka
>  Issue Type: Bug
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
> Fix For: 0.11.0.0
>
>
> Fix findbugs "default character set in use" warnings



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4924) Fix findbugs warnings in Kafka-Connect-API

[ASF GitHub Bot (JIRA)]

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

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

GitHub user cmccabe opened a pull request:

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

KAFKA-4924: Fix Kafka Connect API findbugs warnings



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

$ git pull https://github.com/cmccabe/kafka KAFKA-4924

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

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






> Fix findbugs warnings in Kafka-Connect-API
> --
>
> Key: KAFKA-4924
> URL: https://issues.apache.org/jira/browse/KAFKA-4924
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
>
> Fix findbugs warnings in Kafka-Connect-API



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2715: KAFKA-4924: Fix Kafka Connect API findbugs warning...

[cmccabe]

GitHub user cmccabe opened a pull request:

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

KAFKA-4924: Fix Kafka Connect API findbugs warnings



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

$ git pull https://github.com/cmccabe/kafka KAFKA-4924

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

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






---
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-4924) Fix findbugs warnings in Kafka-Connect-API

[Colin P. McCabe (JIRA)]

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

Colin P. McCabe commented on KAFKA-4924:


* {{connect/api/src/main/java/org/apache/kafka/connect/data/Decimal.java}}: fix 
a warning about a slightly inefficient string conversion.

* 
{{connect/json/src/main/java/org/apache/kafka/connect/json/JsonConverter.java}}:
 fix a potential null pointer exception in an error handling path.

* {{findbugs-exclude.xml}}: add suppressions for an initialization order issue 
in Schema/SchemaBuilder and some unread public fields in the public API

> Fix findbugs warnings in Kafka-Connect-API
> --
>
> Key: KAFKA-4924
> URL: https://issues.apache.org/jira/browse/KAFKA-4924
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Colin P. McCabe
>Assignee: Colin P. McCabe
>
> Fix findbugs warnings in Kafka-Connect-API



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (KAFKA-4924) Fix findbugs warnings in Kafka-Connect-API

[Colin P. McCabe (JIRA)]

Colin P. McCabe created KAFKA-4924:
--

 Summary: Fix findbugs warnings in Kafka-Connect-API
 Key: KAFKA-4924
 URL: https://issues.apache.org/jira/browse/KAFKA-4924
 Project: Kafka
  Issue Type: Bug
  Components: KafkaConnect
Reporter: Colin P. McCabe
Assignee: Colin P. McCabe


Fix findbugs warnings in Kafka-Connect-API



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (KAFKA-4923) Add Exactly-Once Semantics

[Matthias J. Sax (JIRA)]

Matthias J. Sax created KAFKA-4923:
--

 Summary: Add Exactly-Once Semantics
 Key: KAFKA-4923
 URL: https://issues.apache.org/jira/browse/KAFKA-4923
 Project: Kafka
  Issue Type: New Feature
  Components: streams
Reporter: Matthias J. Sax
Assignee: Matthias J. Sax


https://cwiki.apache.org/confluence/display/KAFKA/KIP-129%3A+Streams+Exactly-Once+Semantics



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4862) Kafka client connect to a shutdown node will block for a long time

[Colin P. McCabe (JIRA)]

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

Colin P. McCabe commented on KAFKA-4862:


+1 for adding a configurable connect() timeout for the {{NetworkClient}}.

> Kafka client connect to a shutdown node will block for a long time
> --
>
> Key: KAFKA-4862
> URL: https://issues.apache.org/jira/browse/KAFKA-4862
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.0, 0.10.2.0
>Reporter: Pengwei
>Assignee: Pengwei
> Fix For: 0.11.0.0
>
>
> Currently in our test env, we found after one of the broker node crash(reboot 
> or os crash), the client maybe still connecting to the crash node to send 
> metadata request or other request, and it need about several  minutes to 
> aware the connection is timeout then try another node to connect to send the 
> request.  Then the client may still not aware the metadata change after 
> several minutes.
> We don't have a connection timeout for the network client, we should add a 
> connection timeout for the client



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Re: Re: [DISCUSS] KIP-120: Cleanup Kafka Streams builder API

[Michael Noll]

Hmm, I must admit I don't like this last update all too much.

Basically we would have:

StreamsBuilder builder = new StreamsBuilder();

// And here you'd define your...well, what actually?
// Ah right, you are composing a topology here, though you are not
aware of it.

KafkaStreams streams = new KafkaStreams(builder.build(),
streamsConfiguration);

So what are you building here with StreamsBuilder?  Streams (hint: No)?
And what about tables -- is there a TableBuilder (hint: No)?

I also interpret Guozhang's last response as that he'd prefer to have
"Topology" in the class/interface names.  I am aware that we shouldn't
necessarily use the status quo to make decisions about future changes, but
the very first concept we explain in the Kafka Streams documentation is
"Stream Processing Topology":
https://kafka.apache.org/0102/documentation/streams#streams_concepts

-Michael



On Mon, Mar 20, 2017 at 7:55 PM, Matthias J. Sax 
wrote:

> \cc users list
>
>
>  Forwarded Message 
> Subject: Re: [DISCUSS] KIP-120: Cleanup Kafka Streams builder API
> Date: Mon, 20 Mar 2017 11:51:01 -0700
> From: Matthias J. Sax 
> Organization: Confluent Inc
> To: dev@kafka.apache.org
>
> I want to push this discussion further.
>
> Guozhang's argument about "exposing" the Topology class is valid. It's a
> public class anyway, so it's not as issue. However, I think the question
> is not too much about exposing but about "advertising" (ie, putting it
> into the focus) or not at DSL level.
>
>
> If I interpret the last replies correctly, it seems that we could agree
> on "StreamsBuilder" as name. I did update the KIP accordingly. Please
> correct me, if I got this wrong.
>
>
> If there are not other objects -- this naming discussion was the last
> open point to far -- I would like the start the VOTE thread.
>
>
> -Matthias
>
>
> On 3/14/17 2:37 PM, Guozhang Wang wrote:
> > I'd like to keep the term "Topology" inside the builder class since, as
> > Matthias mentioned, this builder#build() function returns a "Topology"
> > object, whose type is a public class anyways. Although you can argue to
> let
> > users always call
> >
> > "new KafkaStreams(builder.build())"
> >
> > I think it is still more benefit to expose this concept.
> >
> >
> >
> > Guozhang
> >
> > On Tue, Mar 14, 2017 at 10:43 AM, Matthias J. Sax  >
> > wrote:
> >
> >> Thanks for your input Michael.
> >>
>  - KafkaStreams as the new name for the builder that creates the
> logical
>  plan, with e.g. `KafkaStreams.stream("intput-topic")` and
>  `KafkaStreams.table("input-topic")`.
> >>
> >> I don't thinks this is a good idea, for multiple reasons:
> >>
> >> (1) We would reuse a name for a completely different purpose. The same
> >> argument for not renaming KStreamBuilder to TopologyBuilder. The
> >> confusion would just be too large.
> >>
> >> So if we would start from scratch, it might be ok to do so, but now we
> >> cannot make this move, IMHO.
> >>
> >> Also a clarification question: do you suggest to have static methods
> >> #stream and #table -- I am not sure if this would work?
> >> (or was you code snippet just simplification?)
> >>
> >>
> >> (2) Kafka Streams is basically a "processing client" next to consumer
> >> and producer client. Thus, the name KafkaStreams aligns to the naming
> >> schema of KafkaConsumer and KafkaProducer. I am not sure if it would be
> >> a good choice to "break" this naming scheme.
> >>
> >> Btw: this is also the reason, why we have KafkaStreams#close() -- and
> >> not KafkaStreams#stop() -- because #close() aligns with consumer and
> >> producer client.
> >>
> >>
> >> (3) On more argument against using KafkaStreams as DSL entry class would
> >> be, that it would need to create a Topology that can be given to the
> >> "runner/processing-client". Thus the pattern would be
> >>
> >>> Topology topology = streams.build();
> >>> KafkaStramsRunner runner = new KafkaStreamsRunner(..., topology)
> >>
> >> (or of course as a one liner).
> >>
> >>
> >>
> >> On the other hand, there was the idea (that we intentionally excluded
> >> from the KIP), to change the "client instantiation" pattern.
> >>
> >> Right now, a new client in actively instantiated (ie, by calling "new")
> >> and the topology if provided as a constructor argument. However,
> >> especially for DSL (not sure if it would make sense for PAPI), the DSL
> >> builder could create the client for the user.
> >>
> >> Something like this:
> >>
> >>> KStreamBuilder builder = new KStreamBuilder();
> >>> builder.whatever() // use the builder
> >>>
> >>> StreamsConfig config = 
> >>> KafkaStreams streams = builder.getKafkaStreams(config);
> >>
> >> If we change the patter like this, the notion a the "DSL builder" would
> >> change, as it does not create a topology anymore, but it creates the
> >> "processing client". This would address Jay's concern about "not
> >> exposing concept users don't need the understand" and would not require
> >> to include 

Re: [DISCUSS] KIP-120: Cleanup Kafka Streams builder API

[Matthias J. Sax]

I want to push this discussion further.

Guozhang's argument about "exposing" the Topology class is valid. It's a
public class anyway, so it's not as issue. However, I think the question
is not too much about exposing but about "advertising" (ie, putting it
into the focus) or not at DSL level.


If I interpret the last replies correctly, it seems that we could agree
on "StreamsBuilder" as name. I did update the KIP accordingly. Please
correct me, if I got this wrong.


If there are not other objects -- this naming discussion was the last
open point to far -- I would like the start the VOTE thread.


-Matthias


On 3/14/17 2:37 PM, Guozhang Wang wrote:
> I'd like to keep the term "Topology" inside the builder class since, as
> Matthias mentioned, this builder#build() function returns a "Topology"
> object, whose type is a public class anyways. Although you can argue to let
> users always call
> 
> "new KafkaStreams(builder.build())"
> 
> I think it is still more benefit to expose this concept.
> 
> 
> 
> Guozhang
> 
> On Tue, Mar 14, 2017 at 10:43 AM, Matthias J. Sax 
> wrote:
> 
>> Thanks for your input Michael.
>>
 - KafkaStreams as the new name for the builder that creates the logical
 plan, with e.g. `KafkaStreams.stream("intput-topic")` and
 `KafkaStreams.table("input-topic")`.
>>
>> I don't thinks this is a good idea, for multiple reasons:
>>
>> (1) We would reuse a name for a completely different purpose. The same
>> argument for not renaming KStreamBuilder to TopologyBuilder. The
>> confusion would just be too large.
>>
>> So if we would start from scratch, it might be ok to do so, but now we
>> cannot make this move, IMHO.
>>
>> Also a clarification question: do you suggest to have static methods
>> #stream and #table -- I am not sure if this would work?
>> (or was you code snippet just simplification?)
>>
>>
>> (2) Kafka Streams is basically a "processing client" next to consumer
>> and producer client. Thus, the name KafkaStreams aligns to the naming
>> schema of KafkaConsumer and KafkaProducer. I am not sure if it would be
>> a good choice to "break" this naming scheme.
>>
>> Btw: this is also the reason, why we have KafkaStreams#close() -- and
>> not KafkaStreams#stop() -- because #close() aligns with consumer and
>> producer client.
>>
>>
>> (3) On more argument against using KafkaStreams as DSL entry class would
>> be, that it would need to create a Topology that can be given to the
>> "runner/processing-client". Thus the pattern would be
>>
>>> Topology topology = streams.build();
>>> KafkaStramsRunner runner = new KafkaStreamsRunner(..., topology)
>>
>> (or of course as a one liner).
>>
>>
>>
>> On the other hand, there was the idea (that we intentionally excluded
>> from the KIP), to change the "client instantiation" pattern.
>>
>> Right now, a new client in actively instantiated (ie, by calling "new")
>> and the topology if provided as a constructor argument. However,
>> especially for DSL (not sure if it would make sense for PAPI), the DSL
>> builder could create the client for the user.
>>
>> Something like this:
>>
>>> KStreamBuilder builder = new KStreamBuilder();
>>> builder.whatever() // use the builder
>>>
>>> StreamsConfig config = 
>>> KafkaStreams streams = builder.getKafkaStreams(config);
>>
>> If we change the patter like this, the notion a the "DSL builder" would
>> change, as it does not create a topology anymore, but it creates the
>> "processing client". This would address Jay's concern about "not
>> exposing concept users don't need the understand" and would not require
>> to include the word "Topology" in the DSL builder class name, because
>> the builder does not build a Topology anymore.
>>
>> I just put some names that came to my mind first hand -- did not think
>> about good names. It's just to discuss the pattern.
>>
>>
>>
>> -Matthias
>>
>>
>>
>>
>>
>> On 3/14/17 3:36 AM, Michael Noll wrote:
>>> I see Jay's point, and I agree with much of it -- notably about being
>>> careful which concepts we do and do not expose, depending on which user
>>> group / user type is affected.  That said, I'm not sure yet whether or
>> not
>>> we should get rid of "Topology" (or a similar term) in the DSL.
>>>
>>> For what it's worth, here's how related technologies define/name their
>>> "topologies" and "builders".  Note that, in all cases, it's about
>>> constructing a logical processing plan, which then is being executed/run.
>>>
>>> - `Pipeline` (Google Dataflow/Apache Beam)
>>> - To add a source you first instantiate the Source (e.g.
>>> `TextIO.Read.from("gs://some/inputData.txt")`),
>>>   then attach it to your processing plan via
>> `Pipeline#apply()`.
>>>   This setup is a bit different to our DSL because in our DSL the
>>> builder does both, i.e.
>>>   instantiating + auto-attaching to itself.
>>> - To execute the processing plan you call `Pipeline#execute()`.
>>> - `StreamingContext`` (Spark): This setup is similar to our DSL.
>>> - To

[VOTE] KIP-129: Kafka Streams Exactly-Once Semantics

[Matthias J. Sax]

Hi,

I would like to start the vote for KIP-129. Of course, feel free to
provide some more feedback on the DISCUSS thread.

Thanks a lot!


-Matthias



signature.asc
Description: OpenPGP digital signature

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

[Matthias J. Sax]

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

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


-Matthias

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

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

 Marc

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

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

>>>
>>>
>>
> 



signature.asc
Description: OpenPGP digital signature

Re: [DISCUSS] KIP-113: Support replicas movement between log directories

[Dong Lin]

Hey Jun,

Thanks for the response! It seems that we have only two remaining issues.
Please see my reply below.

On Mon, Mar 20, 2017 at 7:45 AM, Jun Rao  wrote:

> Hi, Dong,
>
> Thanks for the update. A few replies inlined below.
>
> On Thu, Mar 16, 2017 at 12:28 AM, Dong Lin  wrote:
>
> > Hey Jun,
> >
> > Thanks for your comment! Please see my reply below.
> >
> > On Wed, Mar 15, 2017 at 9:45 PM, Jun Rao  wrote:
> >
> > > Hi, Dong,
> > >
> > > Thanks for the reply.
> > >
> > > 10. Could you comment on that?
> > >
> >
> > Sorry, I missed that comment.
> >
> > Good point. I think the log segments in topicPartition.move directory
> will
> > be subject to log truncation, log retention and log cleaning in the same
> > way as the log segments in the source log directory. I just specified
> this
> > inthe KIP.
> >
> >
> This is ok, but doubles the overhead of log cleaning. We probably want to
> think a bit more on this.
>

I think this is OK because the number of replicas that are being moved is
limited by the number of ReplicaMoveThread. The default number of
ReplicaMoveThread is the number of log directories, which mean we incur
these overhead for at most one replica per log directory at any time.
Suppose there are most than 100 replica in any log directory, the increase
in overhead is less than 1%.

Another way to look at this is that this is no worse than replica
reassignment. When we reassign replica from one broker to another, we will
double the overhread of log cleaning in the cluster for this replica. If we
are OK with this then we are OK with replica movement between log
directories.


>
>
> >
> > >
> > > 11.2 "I am concerned that the ChangeReplicaDirRequest would be lost if
> > > broker
> > > restarts after it sends ChangeReplicaDirResponse but before it receives
> > > LeaderAndIsrRequest."
> > >
> > > In that case, the reassignment tool could detect that through
> > > DescribeDirsRequest
> > > and issue ChangeReplicaDirRequest again, right? In the common case,
> this
> > is
> > > probably not needed and we only need to write each replica once.
> > >
> > > My main concern with the approach in the current KIP is that once a new
> > > replica is created in the wrong log dir, the cross log directory
> movement
> > > may not catch up until the new replica is fully bootstrapped. So, we
> end
> > up
> > > writing the data for the same replica twice.
> > >
> >
> > I agree with your concern. My main concern is that it is a bit weird if
> > ChangeReplicaDirResponse can not guarantee success and the tool needs to
> > rely on DescribeDirResponse to see if it needs to send
> > ChangeReplicaDirRequest again.
> >
> > How about this: If broker doesn't not have already replica created for
> the
> > specified topicParition when it receives ChangeReplicaDirRequest, it will
> > reply ReplicaNotAvailableException AND remember (replica, destination log
> > directory) pair in memory to create the replica in the specified log
> > directory.
> >
> >
> I am not sure if returning ReplicaNotAvailableException is useful? What
> will the client do on receiving ReplicaNotAvailableException in this case?
>
> Perhaps we could just replace the is_temporary field in
> DescribeDirsRresponsePartition with a state field. We can use 0 to indicate
> the partition is created, 1 to indicate the partition is temporary and 2 to
> indicate that the partition is pending.
>

ReplicaNotAvailableException is useful because the client can re-send
ChangeReplicaDirRequest (with backoff) after receiving
ReplicaNotAvailableException in the response. ChangeReplicaDirRequest will
only succeed after replica has been created for the specified partition in
the broker.

I think this is cleaner than asking reassignment tool to detect that
through DescribeDirsRequest and issue ChangeReplicaDirRequest again. Both
solution has the same chance of writing the data for the same replica
twice. In the original solution, the reassignment tool will keep retrying
ChangeReplicaDirRequest until success. In the second suggested solution,
the reassignment tool needs to send ChangeReplicaDirRequest, send
DescribeDirsRequest to verify result, and retry ChangeReplicaDirRequest and
DescribeDirsRequest again if the replica hasn't been created already. Thus
the second solution couples ChangeReplicaDirRequest with
DescribeDirsRequest and makes tool's logic is bit more complicated.

Besides, I am not sure I understand your suggestion for is_temporary field.
It seems that a replica can have only two states, i.e. normal it is being
used to serve fetch/produce requests and temporary if it is a replica is
that catching up with the normal one. If you think we should have
reassignment tool send DescribeDirsRequest before retrying
ChangeReplicaDirRequest, can you elaborate a bit what is the "pending"
state?


>
>
> > >
> > > 11.3 Are you saying the value in --throttle will be used to set both
> > > intra.broker.throttled.rate and leader.follower.replication.
> > > throttled.replicas?
> > >

[GitHub] kafka pull request #2714: Fix ZKSec Migrate example

[rnpridgeon]

GitHub user rnpridgeon opened a pull request:

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

Fix ZKSec Migrate example

Incorrect option in example 


https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala#L71

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

$ git pull https://github.com/rnpridgeon/kafka patch-1

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

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


commit 2b8d43f4e3760dcda038cb81526f5c66e9c8889c
Author: Ryan P 
Date:   2017-03-20T17:08:02Z

Fix ZKSec Migrate example 

Incorrect option in example 


https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/admin/ZkSecurityMigrator.scala#L71




---
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-4863) Querying window store may return unwanted keys

[ASF GitHub Bot (JIRA)]

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

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

GitHub user dguy opened a pull request:

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

KAFKA-4863: [Follow Up] Querying window store may return unwanted keys

iterate over all keys returned from the rocksdb iterator so we don't miss 
any results

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

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

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

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






> Querying window store may return unwanted keys
> --
>
> Key: KAFKA-4863
> URL: https://issues.apache.org/jira/browse/KAFKA-4863
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Xavier Léauté
>Assignee: Damian Guy
>Priority: Critical
>
> Using variable length keys in a window store may cause unwanted results to be 
> returned when querying certain ranges.
> Below is a test case for {{RocksDBWindowStoreTest}} that shows the problem. 
> It fails, returning {{\[0001, 0003, 0002, 0004, 0005\]}} instead of {{\[0001, 
> 0003, 0005\]}}.
> {code:java}
> @Test
> public void testPutAndFetchSanity() throws IOException {
> final RocksDBWindowStoreSupplier supplier =
> new RocksDBWindowStoreSupplier<>(
> "window", 60 * 1000L * 2, 3,
> true, Serdes.String(), Serdes.String(),
> windowSize, true, Collections.emptyMap(), 
> false
> );
> final WindowStore store = supplier.get();
> store.init(context, store);
> try {
> store.put("a", "0001", 0);
> store.put("aa", "0002", 0);
> store.put("a", "0003", 1);
> store.put("aa", "0004", 1);
> store.put("a", "0005", 6);
> assertEquals(Utils.mkList("0001", "0003", "0005"), 
> toList(store.fetch("a", 0, Long.MAX_VALUE)));
> } finally {
> store.close();
> }
> }
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2713: KAFKA-4863: [Follow Up] Querying window store may ...

[dguy]

GitHub user dguy opened a pull request:

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

KAFKA-4863: [Follow Up] Querying window store may return unwanted keys

iterate over all keys returned from the rocksdb iterator so we don't miss 
any results

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

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

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

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






---
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-4921) AssignedPartition should implement equals

[ASF GitHub Bot (JIRA)]

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

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

GitHub user backender opened a pull request:

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

KAFKA-4921: AssignedPartition implements equals

Solves: [KAFKA-4921](https://issues.apache.org/jira/browse/KAFKA-4921)

Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details)
In class 
org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition
In method 
org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition.compareTo(StreamPartitionAssignor$AssignedPartition)
At StreamPartitionAssignor.java:[line 75]

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

$ git pull https://github.com/delftswa2017/kafka fix-KAFKA-4921

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

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






> AssignedPartition should implement equals
> -
>
> Key: KAFKA-4921
> URL: https://issues.apache.org/jira/browse/KAFKA-4921
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Marc Juchli
>Priority: Minor
>
> We ran FundBug, which resulted in the "Bad practice warning":
>   
> Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
> In class 
> org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition
> In method 
> org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition.compareTo(StreamPartitionAssignor$AssignedPartition)
> At StreamPartitionAssignor.java:[line 75]



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2712: KAFKA-4921: AssignedPartition implements equals

[backender]

GitHub user backender opened a pull request:

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

KAFKA-4921: AssignedPartition implements equals

Solves: [KAFKA-4921](https://issues.apache.org/jira/browse/KAFKA-4921)

Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details)
In class 
org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition
In method 
org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition.compareTo(StreamPartitionAssignor$AssignedPartition)
At StreamPartitionAssignor.java:[line 75]

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

$ git pull https://github.com/delftswa2017/kafka fix-KAFKA-4921

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

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






---
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-4920) Stamped should implement equals

[ASF GitHub Bot (JIRA)]

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

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

GitHub user backender opened a pull request:

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

Stamped implements equals

Solves: [KAFKA-4920](https://issues.apache.org/jira/browse/KAFKA-4920)
Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details)
In class org.apache.kafka.streams.processor.internals.Stamped
In method 
org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]

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

$ git pull https://github.com/delftswa2017/kafka fix-KAFKA-4920

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

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






> Stamped should implement equals
> ---
>
> Key: KAFKA-4920
> URL: https://issues.apache.org/jira/browse/KAFKA-4920
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Marc Juchli
>Priority: Minor
>
> We ran FundBug, which resulted in the "Bad practice warning": 
> Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
> In class org.apache.kafka.streams.processor.internals.Stamped
> In method 
> org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
> At Stamped.java:[lines 31-35]



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2711: Stamped implements equals

[backender]

GitHub user backender opened a pull request:

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

Stamped implements equals

Solves: [KAFKA-4920](https://issues.apache.org/jira/browse/KAFKA-4920)
Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details)
In class org.apache.kafka.streams.processor.internals.Stamped
In method 
org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]

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

$ git pull https://github.com/delftswa2017/kafka fix-KAFKA-4920

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

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






---
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.
---

[VOTE] KIP-124: Request rate quotas

[Rajini Sivaram]

I would like to initiate the voting process for KIP-124:

https://cwiki.apache.org/confluence/display/KAFKA/KIP-124+-+Request+rate+quotas

The KIP proposes to add request processing time quota to limit CPU
utilization at **, ** or ** levels
similar to the current bandwidth limiting quotas. The quota takes into
account the total time spent by the user/client on request handler and
network threads within a quota window. Quota will be configured as a
per-thread percentage value, which typically indicates the percentage of
CPU cores allocated to the user/client.

The discussion thread is here:


https://lists.apache.org/thread.html/46c7bbc8f381ebe718b3cce6ed8bdf3745df22b0bd88020d70c99813@%3Cdev.kafka.apache.org%3E

Many thanks to everyone for the feedback and suggestions so far.

Regards,

Rajini

Re: [DISCUSS] KIP-113: Support replicas movement between log directories

[Jun Rao]

Hi, Dong,

Thanks for the update. A few replies inlined below.

On Thu, Mar 16, 2017 at 12:28 AM, Dong Lin  wrote:

> Hey Jun,
>
> Thanks for your comment! Please see my reply below.
>
> On Wed, Mar 15, 2017 at 9:45 PM, Jun Rao  wrote:
>
> > Hi, Dong,
> >
> > Thanks for the reply.
> >
> > 10. Could you comment on that?
> >
>
> Sorry, I missed that comment.
>
> Good point. I think the log segments in topicPartition.move directory will
> be subject to log truncation, log retention and log cleaning in the same
> way as the log segments in the source log directory. I just specified this
> inthe KIP.
>
>
This is ok, but doubles the overhead of log cleaning. We probably want to
think a bit more on this.


>
> >
> > 11.2 "I am concerned that the ChangeReplicaDirRequest would be lost if
> > broker
> > restarts after it sends ChangeReplicaDirResponse but before it receives
> > LeaderAndIsrRequest."
> >
> > In that case, the reassignment tool could detect that through
> > DescribeDirsRequest
> > and issue ChangeReplicaDirRequest again, right? In the common case, this
> is
> > probably not needed and we only need to write each replica once.
> >
> > My main concern with the approach in the current KIP is that once a new
> > replica is created in the wrong log dir, the cross log directory movement
> > may not catch up until the new replica is fully bootstrapped. So, we end
> up
> > writing the data for the same replica twice.
> >
>
> I agree with your concern. My main concern is that it is a bit weird if
> ChangeReplicaDirResponse can not guarantee success and the tool needs to
> rely on DescribeDirResponse to see if it needs to send
> ChangeReplicaDirRequest again.
>
> How about this: If broker doesn't not have already replica created for the
> specified topicParition when it receives ChangeReplicaDirRequest, it will
> reply ReplicaNotAvailableException AND remember (replica, destination log
> directory) pair in memory to create the replica in the specified log
> directory.
>
>
I am not sure if returning ReplicaNotAvailableException is useful? What
will the client do on receiving ReplicaNotAvailableException in this case?

Perhaps we could just replace the is_temporary field in
DescribeDirsRresponsePartition with a state field. We can use 0 to indicate
the partition is created, 1 to indicate the partition is temporary and 2 to
indicate that the partition is pending.


> >
> > 11.3 Are you saying the value in --throttle will be used to set both
> > intra.broker.throttled.rate and leader.follower.replication.
> > throttled.replicas?
> >
>
> No. --throttle will be used to only to set leader.follower.replication as
> it does now. I think we do not need any option in the
> kafka-reassignment-partitions.sh to specify intra.broker.throttled.rate.
> User canset it in broker config or dynamically using kafka-config.sh. Does
> this sound OK?
>
>
Ok. This sounds good. It would be useful to make this clear in the wiki.


>
> >
> > 12.2 If the user only wants to check one topic, the tool could do the
> > filtering on the client side, right? My concern with having both log_dirs
> > and topics is the semantic. For example, if both are not empty, do we
> > return the intersection or the union?
> >
>
> Yes the tool could filter on the client side. But the purpose of having
> this field is to reduce response side in case broker has a lot of topics.
> The both fields are used as filter and the result is intersection. Do you
> think this semantic is confusing or counter-intuitive?


>

Ok. Could we document the semantic when both dirs and topics are specified?

Thanks,

Jun

>
> >
> > On Mon, Mar 13, 2017 at 3:32 PM, Dong Lin  wrote:
> >
> > > Hey Jun,
> > >
> > > Thanks much for your detailed comments. Please see my reply below.
> > >
> > > On Mon, Mar 13, 2017 at 9:09 AM, Jun Rao  wrote:
> > >
> > > > Hi, Dong,
> > > >
> > > > Thanks for the updated KIP. Some more comments below.
> > > >
> > > > 10. For the .move log, do we perform any segment deletion (based on
> > > > retention) or log cleaning (if a compacted topic)? Or do we only
> enable
> > > > that after the swap?
> > > >
> > > > 11. kafka-reassign-partitions.sh
> > > > 11.1 If all reassigned replicas are in the current broker and only
> the
> > > log
> > > > directories have changed, we can probably optimize the tool to not
> > > trigger
> > > > partition reassignment through the controller and only
> > > > send ChangeReplicaDirRequest.
> > > >
> > >
> > > Yes, the reassignment script should not create the reassignment znode
> if
> > no
> > > replicas are not be moved between brokers. This falls into the "How to
> > move
> > > replica between log directories on the same broker" of the Proposed
> > Change
> > > section.
> > >
> > >
> > > > 11.2 If ChangeReplicaDirRequest specifies a replica that's not
> created
> > > yet,
> > > > could the broker just remember that in memory and create the replica
> > when
> > > > the creation is requested? This way, when doing cluster expansion

Re: [VOTE] KIP-111 Kafka should preserve the Principal generated by the PrincipalBuilder while processing the request received on socket channel, on the broker.

[Jun Rao]

Hi, Mayuresh,

One reason to have KafkaPrincipal in ACL is that we can extend it to
support group in the future. Have you thought about how to support that in
your new proposal?

Another reason that we had KafkaPrincipal is simplicity. It can be
constructed from a simple string and makes matching easier. If we
expose java.security.Principal,
then I guess that when an ACL is set, we have to be able to construct
a java.security.Principal
from some string to match the java.security.Principal generated from the
SSL or SASL library. How do we make sure that the same type of
java.security.Principal
can be created and will match?

Thanks,

Jun


On Wed, Mar 15, 2017 at 8:48 PM, Mayuresh Gharat  wrote:

> Hi Jun,
>
> Sorry for the delayed reply.
> I agree that the easiest thing will be to add an additional field in the
> Session class and we should be OK.
> But having a KafkaPrincipal and java Principal with in the same class looks
> little weird.
>
> So we can do this and slowly deprecate the usage of KafkaPrincipal in
> public api's.
>
> We add new apis and make changes to the existing apis as follows :
>
>
>- Changes to Session class :
>
> @Deprecated
> case class Session(principal: KafkaPrincipal, clientAddress: InetAddress) {
> val sanitizedUser = QuotaId.sanitize(principal.getName)
> }
>
>
> *@Deprecated .. (NEW)*
>
>
> *case class Session(principal: KafkaPrincipal, clientAddress: InetAddress,
> channelPrincipal: Java.security.Principal) {val sanitizedUser =
> QuotaId.sanitize(principal.getName)}*
>
> *(NEW)*
>
>
> *case class Session(principal: Java.security.Principal, clientAddress:
> InetAddress) {val sanitizedUser = QuotaId.sanitize(principal.get
> Name)}*
>
>
>- Changes to Authorizer Interface :
>
> @Deprecated
> def getAcls(principal: KafkaPrincipal): Map[Resource, Set[Acl]]
>
> *(NEW)*
> *def getAcls(principal: Java.security.Principal): Map[Resource, Set[Acl]]*
>
>
>- Changes to Acl class :
>
> @Deprecated
> case class Acl(principal: KafkaPrincipal, permissionType: PermissionType,
> host: String, operation: Operation)
>
>*(NEW)*
>
>
> *case class Acl(principal: Java.security.Principal, permissionType:
> PermissionType, host: String, operation: Operation) *
> The one in Bold are the new api's. We will remove them eventually, probably
> in next major release.
> We don't want to get rid of KafkaPrincipal class and it will be used in the
> same way as it does right now for out of box authorizer and commandline
> tool. We would only be removing its direct usage from public apis.
> Doing the above deprecation will help us to support other implementation of
> Java.security.Principal as well which seems necessary especially since
> Kafka provides pluggable Authorizer and PrincipalBuilder.
>
> Let me know your thoughts on this.
>
> Thanks,
>
> Mayuresh
>
> On Tue, Feb 28, 2017 at 2:33 PM, Mayuresh Gharat <
> gharatmayures...@gmail.com
> > wrote:
>
> > Hi Jun,
> >
> > Sure.
> > I had an offline discussion with Joel on how we can deprecate the
> > KafkaPrincipal from  Session and Authorizer.
> > I will update the KIP to see if we can address all the concerns here. If
> > not we can keep the KafkaPrincipal.
> >
> > Thanks,
> >
> > Mayuresh
> >
> > On Tue, Feb 28, 2017 at 1:53 PM, Jun Rao  wrote:
> >
> >> Hi, Joel,
> >>
> >> Good point on the getAcls() method. KafkaPrincipal is also tied to ACL,
> >> which is used in pretty much every method in Authorizer. Now, I am not
> >> sure
> >> if it's easy to deprecate KafkaPrincipal.
> >>
> >> Hi, Mayuresh,
> >>
> >> Given the above, it seems that the easiest thing is to add a new
> Principal
> >> field in Session. We want to make it clear that it's ignored in the
> >> default
> >> implementation, but a customizer authorizer could take advantage of
> that.
> >>
> >> Thanks,
> >>
> >> Jun
> >>
> >> On Tue, Feb 28, 2017 at 10:52 AM, Joel Koshy 
> wrote:
> >>
> >> > If we deprecate KafkaPrincipal, then the Authorizer interface will
> also
> >> > need to change - i.e., deprecate the getAcls(KafkaPrincipal) method.
> >> >
> >> > On Tue, Feb 28, 2017 at 10:11 AM, Mayuresh Gharat <
> >> > gharatmayures...@gmail.com> wrote:
> >> >
> >> > > Hi Jun/Ismael,
> >> > >
> >> > > Thanks for the comments.
> >> > >
> >> > > I agree.
> >> > > What I was thinking was, we get the KIP passed now and wait till
> major
> >> > > kafka version release. We can then make this change, but for now we
> >> can
> >> > > wait. Does that work?
> >> > >
> >> > > If there are concerns, we can make the addition of extra field of
> type
> >> > > Principal to Session and then deprecate the KafkaPrincipal later.
> >> > >
> >> > > I am fine either ways. What do you think?
> >> > >
> >> > > Thanks,
> >> > >
> >> > > Mayuresh
> >> > >
> >> > > On Tue, Feb 28, 2017 at 9:53 AM, Jun Rao  wrote:
> >> > >
> >> > > > Hi, Ismael,
> >> > > >
> >> > > > Good point on compatibility.
> >> > > >
> >> > > > Hi, Mayuresh,
> >> > > >
> >> > > > Given that, it seems that it's better to just

[jira] [Updated] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

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

Daan Rennings updated KAFKA-4922:
-
Status: Patch Available  (was: Open)

> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

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

Daan Rennings updated KAFKA-4922:
-
Flags:   (was: Patch)

> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

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

Daan Rennings updated KAFKA-4922:
-
Flags: Patch

> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4890) State directory being deleted when another thread holds the lock

[Yunus Olgun (JIRA)]

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

Yunus Olgun commented on KAFKA-4890:


Happy to help, but I wasn't so helpful this time, unfortunately.

Please disregard the unit test. Calling _get_ immediately after _submit_ turned 
this supposedly multithreaded test into a sequential one. Couldn't reproduce 
using proper Futures and simulated wait times. Also 
OverlappingFileLockException should protect against multithreaded or single 
threaded access from within same VM. My assumption was wrong.

In logs3.tar.gz file, timestamps are in seconds and logs are coming from 
different threads. So, order of logs between different threads may not be 
correct. I will try to reproduce the issue using milliseconds and AsyncAppender 
in log configuration.

- In 0.10.2.0, this bug was a blocker for me to use multithreaded in streams 
application. It happens frequently, rebalancing and state store initialization 
takes too long.
- In 0.11.0.0, with default configuration I couldn't reproduce it. Even with 
state.cleanup.delay.ms=100, it takes some time. Also rebalancing and state 
store initalization is much faster now. It is not urgent for this version, imo.

> State directory being deleted when another thread holds the lock
> 
>
> Key: KAFKA-4890
> URL: https://issues.apache.org/jira/browse/KAFKA-4890
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Damian Guy
> Attachments: logs2.tar.gz, logs3.tar.gz, logs.tar.gz
>
>
> Looks like a state directory is being cleaned up when another thread already 
> has the lock:
> {code}
> 2017-03-12 20:39:17 [StreamThread-1] DEBUG o.a.k.s.p.i.ProcessorStateManager 
> - task [0_6] Registering state store perGameScoreStore to its state manager
> 2017-03-12 20:40:21 [StreamThread-3] INFO  o.a.k.s.p.i.StateDirectory - 
> Deleting obsolete state directory 0_6 for task 0_6
> 2017-03-12 20:40:22 [StreamThread-1] ERROR o.a.k.c.c.i.ConsumerCoordinator - 
> User provided listener 
> org.apache.kafka.streams.processor.internals.StreamThread$1 for group 
> fireflyProd failed on partition assignment
> org.apache.kafka.streams.errors.ProcessorStateException: Error while 
> executing put key 
> \x00\x00\x00\x00}\xA2\x9E\x9D\x05\xF6\x95\xAB\x01\x12dayOfGame and value 
> \x00\x00\x00\x00z\x00\x00\x00\x00\x00\x80G@ from store perGameScoreStore
> at 
> org.apache.kafka.streams.state.internals.RocksDBStore.putInternal(RocksDBStore.java:248)
> at 
> org.apache.kafka.streams.state.internals.RocksDBStore.access$000(RocksDBStore.java:65)
> at 
> org.apache.kafka.streams.state.internals.RocksDBStore$1.restore(RocksDBStore.java:156)
> at 
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.restoreActiveState(ProcessorStateManager.java:230)
> at 
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:193)
> at 
> org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:99)
> at 
> org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:152)
> at 
> org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.init(ChangeLoggingKeyValueBytesStore.java:39)
> at 
> org.apache.kafka.streams.state.internals.MeteredKeyValueStore$7.run(MeteredKeyValueStore.java:100)
> at 
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
> at 
> org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:131)
> at 
> org.apache.kafka.streams.state.internals.CachingKeyValueStore.init(CachingKeyValueStore.java:62)
> at 
> org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:86)
> at 
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:141)
> at 
> org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:834)
> at 
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:1207)
> at 
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:1180)
> at 
> org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:937)
> at 
> org.apache.kafka.streams.processor.internals.StreamThread.access$500(StreamThread.java:69)
>  at 
> org.apache.kafka.streams.processor.internals.StreamThread$1.onPartitionsAssigned(StreamThread.java:236)
> at 
> org.apache.kafka.cl

[jira] [Commented] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[ASF GitHub Bot (JIRA)]

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

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

GitHub user drennings opened a pull request:

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

Fix-KAFKA-4922 - Minor FindBugs warning fixes

Minor one line adaptations to one bad practice and three performance 
warnings
https://issues.apache.org/jira/browse/KAFKA-4922

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

$ git pull https://github.com/delftswa2017/kafka fix-KAFKA-4922

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

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


commit 5fcc4584d2f325034eb1e6e8708bd7db2dabb08b
Author: Steven Schlansker 
Date:   2017-02-16T05:19:15Z

KAFKA-2740: Add a KStream#peek(ForeachAction) in DSL

https://issues.apache.org/jira/browse/KAFKA-4720

Peek is a handy method to have to insert diagnostics that do not affect the 
stream itself, but some external state such as logging or metrics collection.

Author: Steven Schlansker 

Reviewers: Damian Guy, Matthias J. Sax, Eno Thereska, Guozhang Wang

Closes #2493 from stevenschlansker/kafka-4720-peek

commit 6a6f488108a5b89f694f2c40b2d18b279c248d85
Author: daan rennings 
Date:   2017-03-20T13:37:19Z

Fixed adherence to java.lang.Object.equals in Bytes.java

commit a4c83ec0b286ddc813d13951b57cccda6e5a4d98
Author: daan rennings 
Date:   2017-03-20T13:43:48Z

Fixed boxed primitive in Decimal.java

commit 92342a953c006a997180ebb6289249e390dc8c47
Author: daan rennings 
Date:   2017-03-20T13:48:21Z

Fixed two boxed primitives in Worker.java

commit 6c38737049c4eaedfad3cfb87fc227825402f315
Author: daan rennings 
Date:   2017-03-20T13:51:08Z

Fixed boxed primitive in ConnectorTaskId.java




> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

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

[Damian Guy]

Hi Marc,

Thanks for the KIP. It mostly looks good to me. The only thing i'd change
is using a null argument to use a default mapping. IMO it would be better
if the existing print() method delegates to the new one supplying a
KeyValueMapper that does the right thing.

Thanks,
Damian

On Sat, 18 Mar 2017 at 14:25 Marc Juchli  wrote:

> Thanks!
>
> I wanted to PING this thread. Not sure what the next steps of the KIP
> process are?
>
> Kind regards,
> Marc
>
> On Wed, Mar 15, 2017 at 9:13 PM Matthias J. Sax 
> wrote:
>
> > Thanks for updating the KIP.
> >
> > It's in very good shape IMHO and I support this idea!
> >
> >
> >
> > -Matthias
> >
> >
> > On 3/15/17 3:05 AM, Marc Juchli wrote:
> > > Dear Matthias,
> > >
> > > The KIP is updated. I think it now contains all the information on that
> > > page.
> > >
> > > Marc
> > >
> > > On Mon, Mar 13, 2017 at 9:37 PM Matthias J. Sax  >
> > > wrote:
> > >
> > >> Marc,
> > >>
> > >> Thanks for the KIP.
> > >>
> > >> Can you please update the KIP in a way such that it is self contained.
> > >> Right now, you link to all kind of other places making it hard to read
> > >> the KIP.
> > >>
> > >> The KIP should be the "center of truth" -- if there is important
> > >> information elsewhere, please c&p it into the KIP.
> > >>
> > >>
> > >> Thanks a lot!
> > >>
> > >>
> > >> -Matthias
> > >>
> > >>
> > >>
> > >> On 3/13/17 1:30 PM, Matthias J. Sax wrote:
> > >>> Can you please add the KIP to this table:
> > >>>
> > >>>
> > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Improvement+Proposals#KafkaImprovementProposals-KIPsunderdiscussion
> > >>>
> > >>> Thanks,
> > >>>
> > >>>  Matthias
> > >>>
> > >>>
> > >>> On 3/13/17 8:08 AM, Marc Juchli wrote:
> >  Dear all,
> > 
> >  The following describes KIP-132, which I just created. See:
> > 
> > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-132+-+Augment+KStream.print+to+allow+extra+parameters+in+the+printed+string
> > 
> >  Motivation
> > 
> >  As for now, KStream#print leads to a predefined output where key and
> > >> value are
> >  printed with comma separation.
> >  KAFKA-4830 
> > suggests
> > >> to
> >  extend print in a way that it takes KeyValueMapper as a parameter.
> >  This will allow a user to change outputs according to the users
> > demand.
> >  Public Interfaces
> > 
> >  The affected interface is KStream, which needs to be extended with
> > >> another
> >  overloaded version of print:
> > 
> >  void print(final Serde keySerde,
> > final Serde valSerde,
> > final String streamName,
> > final KeyValueMapper mapper);
> > 
> >  Proposed Changes
> > 
> >  See pull request GH-2669  >.
> >  This PR contains a discussion regarding KAFKA-4830
> >   as well as
> > >> KAFKA-4772
> >  .
> > 
> >  Compatibility, Deprecation, and Migration Plan
> > 
> >  The extension of print will not introduce compatibility issues – we
> > can
> >  maintain the current output by keeping the current output format as
> a
> >  default (if mapper was not set):
> > 
> >  if(mapper == null) {
> >  printStream.println("[" + streamName + "]: " + keyToPrint + " ,
> "
> >  + valueToPrint);
> >  } else {
> >  printStream.println("[" + streamName + "]: " +
> >  mapper.apply(keyToPrint, valueToPrint));
> >  }
> > 
> > 
> > 
> >  Kind regards,
> >  Marc
> > 
> > >>>
> > >>
> > >>
> > >
> >
> >
>

[GitHub] kafka pull request #2710: Fix-KAFKA-4922 - Minor FindBugs warning fixes

[drennings]

GitHub user drennings opened a pull request:

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

Fix-KAFKA-4922 - Minor FindBugs warning fixes

Minor one line adaptations to one bad practice and three performance 
warnings
https://issues.apache.org/jira/browse/KAFKA-4922

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

$ git pull https://github.com/delftswa2017/kafka fix-KAFKA-4922

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

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


commit 5fcc4584d2f325034eb1e6e8708bd7db2dabb08b
Author: Steven Schlansker 
Date:   2017-02-16T05:19:15Z

KAFKA-2740: Add a KStream#peek(ForeachAction) in DSL

https://issues.apache.org/jira/browse/KAFKA-4720

Peek is a handy method to have to insert diagnostics that do not affect the 
stream itself, but some external state such as logging or metrics collection.

Author: Steven Schlansker 

Reviewers: Damian Guy, Matthias J. Sax, Eno Thereska, Guozhang Wang

Closes #2493 from stevenschlansker/kafka-4720-peek

commit 6a6f488108a5b89f694f2c40b2d18b279c248d85
Author: daan rennings 
Date:   2017-03-20T13:37:19Z

Fixed adherence to java.lang.Object.equals in Bytes.java

commit a4c83ec0b286ddc813d13951b57cccda6e5a4d98
Author: daan rennings 
Date:   2017-03-20T13:43:48Z

Fixed boxed primitive in Decimal.java

commit 92342a953c006a997180ebb6289249e390dc8c47
Author: daan rennings 
Date:   2017-03-20T13:48:21Z

Fixed two boxed primitives in Worker.java

commit 6c38737049c4eaedfad3cfb87fc227825402f315
Author: daan rennings 
Date:   2017-03-20T13:51:08Z

Fixed boxed primitive in ConnectorTaskId.java




---
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-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

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

Daan Rennings commented on KAFKA-4922:
--

I would like to tackle this issue myself. However I am not sure how issues are 
assigned.

> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

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

Daan Rennings updated KAFKA-4922:
-
Reviewer:   (was: Daan Rennings)

> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

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

Daan Rennings updated KAFKA-4922:
-
Reviewer: Daan Rennings

> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

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

[Michael Pearce]

Hi Ismael, 

Sorry, 

The response below was in regards to your comments, got my wires crossed, 
apologies.

Hi Jun,

I’m happy with the change, I see Jason updated our KIP, many thanks for this, 
and thanks for implementing for us ☺

Cheers
Mike

On 20/03/2017, 13:19, "Michael Pearce"  wrote:

Hi Jun,

Thanks the comments I’ve updated the KIP a little where agreement.

My comments:

1) Good point, removed from the interface. See updated KIP
2) I think, Radai’s suggested header(String key) is a cleaner method name, 
but happy to change if community believe lastHeader is better. I’ll keep 
Radai’s suggested name for now.
3) Agreed, added IllegalStateException to interface. See updated KIP
4) I think for now intent would be to simply instantiate a new object from 
the provided implementation that would exist and implement the Header interface 
e.g. I would expect user to invoke “new HeaderRecord(String key, byte[] value)” 
for now.
5) Agreed, we should add later if useful. I think as per other arguments of 
trying to keep the methods for now limited as easier to add new methods later, 
but cannot take away, as such I think we should avoid adding this, as you note 
it is still possible with the current suggested api.

Cheers
Mike

On 17/03/2017, 13:47, "isma...@gmail.com on behalf of Ismael Juma" 
 wrote:

Jun, the message format you proposed seems reasonable to me. I have a 
few
minor comments with regards to the user facing API:

1. Do we want to expose the `close()` method in the Headers interface? 
It
seems that this method should only be called by the producer after the
headers have been passed to the interceptors, so it may make sense to 
keep
it as an internal method in the implementation class.

2. `header(String key)` returns the last header for that key. Maybe we
should make it explicit by calling the method `lastHeader(String key)`.

3. I agree with the change to throw an exception if we try to modify the
headers when they are in read-only mode. We should specify the 
exception in
the KIP. IllegalStateException, as suggested by Radai, seems reasonable.

4. How do users create a `Header` instance to pass to the `add` method? 
We
could introduce a static `create` method that takes both parameters to 
the
`Header` interface (requires Java 8).

5. There's no method to replace all the headers with a given key so one
would have to call `remove` and then `add`. Is the assumption that this 
is
rare? If so, that's probably OK, we can add another method later, if 
it's
useful.

Thanks,
Ismael

On Thu, Mar 16, 2017 at 4:44 PM, Jun Rao  wrote:

> Hi, Everyone,
>
> Jason has been working on the new message format related to EOS (
> https://github.com/apache/kafka/pull/2614). He has included the header
> changes proposed in the KIP, which reduces the overhead for 
supporting an
> additional message format change if done separately. Since the message
> format part of the header proposal seems less controversial and the
> consensus is header is needed, does anyone have objections to this? 
The
> following is the new record format with headers.
>
> * Record =>
> *   Length => Varint
> *   Attributes => Int8
> *   TimestampDelta => Varlong
> *   OffsetDelta => Varint
> *   Key => Bytes
> *   Value => Bytes
> *   Headers => [HeaderKey HeaderValue]
> * HeaderKey => String
> * HeaderValue => Bytes
> *
> * Note that in this schema, the Bytes and String types use a variable
> length integer to represent
> * the length of the field. The array type used for the headers also
> uses a Varint for the number of
> * headers.
>
> Thanks,
>
> Jun
>
>
> On Tue, Mar 14, 2017 at 10:49 AM, Ismael Juma  
wrote:
>
> > Thanks Radai. Great to have a concrete example of the intended 
usage.
> >
> > Regarding performance, we would need to benchmark, as you said. But 
there
> > would be a lot of reuse (in essence, we are copying 5 references 
plus a
> new
> > object header), so I'd be surprised if that would be the bottleneck
> > compared to some of the other allocations that would be happening 
in that
> > path. In any case, I think we can leave this aside for now since 
people
> > also felt that the mutable API would be easier to use.
> >
> > About ProducerRecord reuse, my understanding is that people do 
sometimes
> > retry a failed request manually due to the fact that a large retry 
number
> > d

[jira] [Created] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

Daan Rennings created KAFKA-4922:


 Summary: Fix several FindBugs warnings in Clients and Connect
 Key: KAFKA-4922
 URL: https://issues.apache.org/jira/browse/KAFKA-4922
 Project: Kafka
  Issue Type: Improvement
  Components: clients, KafkaConnect
Affects Versions: 0.10.2.0
 Environment: Identified by FindBugs, non-software and -platform 
specific
Reporter: Daan Rennings
Priority: Minor
 Fix For: 0.10.2.1
 Attachments: ClientsFindBugsReport.html, 
ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html

Three easy to fix warnings (not a complete set of the current FindBugs 
warnings) as identified by FindBugs and stated in the attached reports:
-org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
argument (Low priority, Bad Practice)
-Primitive boxed just to call toString in 
org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, Performance)
-Primitive boxed just to call toString in new 
org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
WorkerConfig, OffsetBackingStore) (High Priority, Performance)
-Primitive is boxed to call Integer.compareTo(Integer): use 
Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4922) Fix several FindBugs warnings in Clients and Connect

[Daan Rennings (JIRA)]

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

Daan Rennings updated KAFKA-4922:
-
Description: 
Four easy to fix warnings (not a complete set of the current FindBugs warnings) 
as identified by FindBugs and stated in the attached reports:
-org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
argument (Low priority, Bad Practice)
-Primitive boxed just to call toString in 
org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, Performance)
-Primitive boxed just to call toString in new 
org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
WorkerConfig, OffsetBackingStore) (High Priority, Performance)
-Primitive is boxed to call Integer.compareTo(Integer): use 
Integer.compare(int, int) instead (High Priority, Performance)

  was:
Three easy to fix warnings (not a complete set of the current FindBugs 
warnings) as identified by FindBugs and stated in the attached reports:
-org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
argument (Low priority, Bad Practice)
-Primitive boxed just to call toString in 
org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, Performance)
-Primitive boxed just to call toString in new 
org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
WorkerConfig, OffsetBackingStore) (High Priority, Performance)
-Primitive is boxed to call Integer.compareTo(Integer): use 
Integer.compare(int, int) instead (High Priority, Performance)


> Fix several FindBugs warnings in Clients and Connect
> 
>
> Key: KAFKA-4922
> URL: https://issues.apache.org/jira/browse/KAFKA-4922
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, KafkaConnect
>Affects Versions: 0.10.2.0
> Environment: Identified by FindBugs, non-software and -platform 
> specific
>Reporter: Daan Rennings
>Priority: Minor
>  Labels: newbie
> Fix For: 0.10.2.1
>
> Attachments: ClientsFindBugsReport.html, 
> ConnectAPIFindBugsReport.html, ConnectRuntimeFindBugsReport.html
>
>
> Four easy to fix warnings (not a complete set of the current FindBugs 
> warnings) as identified by FindBugs and stated in the attached reports:
> -org.apache.kafka.common.utils.Bytes.equals(Object) does not check for null 
> argument (Low priority, Bad Practice)
> -Primitive boxed just to call toString in 
> org.apache.kafka.connect.data.Decimal.builder(int) (High Priority, 
> Performance)
> -Primitive boxed just to call toString in new 
> org.apache.kafka.connect.runtime.Worker(String, Time, ConnectorFactory, 
> WorkerConfig, OffsetBackingStore) (High Priority, Performance)
> -Primitive is boxed to call Integer.compareTo(Integer): use 
> Integer.compare(int, int) instead (High Priority, Performance)



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

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

[Michael Pearce]

Hi Jun,

Thanks the comments I’ve updated the KIP a little where agreement.

My comments:

1) Good point, removed from the interface. See updated KIP
2) I think, Radai’s suggested header(String key) is a cleaner method name, but 
happy to change if community believe lastHeader is better. I’ll keep Radai’s 
suggested name for now.
3) Agreed, added IllegalStateException to interface. See updated KIP
4) I think for now intent would be to simply instantiate a new object from the 
provided implementation that would exist and implement the Header interface 
e.g. I would expect user to invoke “new HeaderRecord(String key, byte[] value)” 
for now.
5) Agreed, we should add later if useful. I think as per other arguments of 
trying to keep the methods for now limited as easier to add new methods later, 
but cannot take away, as such I think we should avoid adding this, as you note 
it is still possible with the current suggested api.

Cheers
Mike

On 17/03/2017, 13:47, "isma...@gmail.com on behalf of Ismael Juma" 
 wrote:

Jun, the message format you proposed seems reasonable to me. I have a few
minor comments with regards to the user facing API:

1. Do we want to expose the `close()` method in the Headers interface? It
seems that this method should only be called by the producer after the
headers have been passed to the interceptors, so it may make sense to keep
it as an internal method in the implementation class.

2. `header(String key)` returns the last header for that key. Maybe we
should make it explicit by calling the method `lastHeader(String key)`.

3. I agree with the change to throw an exception if we try to modify the
headers when they are in read-only mode. We should specify the exception in
the KIP. IllegalStateException, as suggested by Radai, seems reasonable.

4. How do users create a `Header` instance to pass to the `add` method? We
could introduce a static `create` method that takes both parameters to the
`Header` interface (requires Java 8).

5. There's no method to replace all the headers with a given key so one
would have to call `remove` and then `add`. Is the assumption that this is
rare? If so, that's probably OK, we can add another method later, if it's
useful.

Thanks,
Ismael

On Thu, Mar 16, 2017 at 4:44 PM, Jun Rao  wrote:

> Hi, Everyone,
>
> Jason has been working on the new message format related to EOS (
> https://github.com/apache/kafka/pull/2614). He has included the header
> changes proposed in the KIP, which reduces the overhead for supporting an
> additional message format change if done separately. Since the message
> format part of the header proposal seems less controversial and the
> consensus is header is needed, does anyone have objections to this? The
> following is the new record format with headers.
>
> * Record =>
> *   Length => Varint
> *   Attributes => Int8
> *   TimestampDelta => Varlong
> *   OffsetDelta => Varint
> *   Key => Bytes
> *   Value => Bytes
> *   Headers => [HeaderKey HeaderValue]
> * HeaderKey => String
> * HeaderValue => Bytes
> *
> * Note that in this schema, the Bytes and String types use a variable
> length integer to represent
> * the length of the field. The array type used for the headers also
> uses a Varint for the number of
> * headers.
>
> Thanks,
>
> Jun
>
>
> On Tue, Mar 14, 2017 at 10:49 AM, Ismael Juma  wrote:
>
> > Thanks Radai. Great to have a concrete example of the intended usage.
> >
> > Regarding performance, we would need to benchmark, as you said. But 
there
> > would be a lot of reuse (in essence, we are copying 5 references plus a
> new
> > object header), so I'd be surprised if that would be the bottleneck
> > compared to some of the other allocations that would be happening in 
that
> > path. In any case, I think we can leave this aside for now since people
> > also felt that the mutable API would be easier to use.
> >
> > About ProducerRecord reuse, my understanding is that people do sometimes
> > retry a failed request manually due to the fact that a large retry 
number
> > doesn't help if a batch is expired in the queue. I believe KIP-91 will
> > help:
> >
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > 91+Provide+Intuitive+User+Timeouts+in+The+Producer
> >
> > In addition, KIP-98 (Exactly-once) won't achieve its goal if people do
> > manual retries. So, it seems like it's OK to require people to create a
> new
> > ProducerRecord if they really need to do manual retries. But we should
> add
> > a note to the compatibility section of the KIP.
> >
> > I have a few minor API suggestions. I'll send a follow-up later today,
> > hopefully.
> >
> > Ismael
> >
> > On 

[jira] [Created] (KAFKA-4921) AssignedPartition should implement equals

[Marc Juchli (JIRA)]

Marc Juchli created KAFKA-4921:
--

 Summary: AssignedPartition should implement equals
 Key: KAFKA-4921
 URL: https://issues.apache.org/jira/browse/KAFKA-4921
 Project: Kafka
  Issue Type: Improvement
  Components: streams
Reporter: Marc Juchli
Priority: Minor


We ran FundBug, which resulted in the "Bad practice warning":


Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class 
org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition
In method 
org.apache.kafka.streams.processor.internals.StreamPartitionAssignor$AssignedPartition.compareTo(StreamPartitionAssignor$AssignedPartition)
At StreamPartitionAssignor.java:[line 75]



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (KAFKA-4920) Stamped should implement equals

[Marc Juchli (JIRA)]

Marc Juchli created KAFKA-4920:
--

 Summary: Stamped should implement equals
 Key: KAFKA-4920
 URL: https://issues.apache.org/jira/browse/KAFKA-4920
 Project: Kafka
  Issue Type: Improvement
  Components: streams
Reporter: Marc Juchli
Priority: Minor


We ran FundBug, which resulted in the "Bad practice warning": 
Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class org.apache.kafka.streams.processor.internals.Stamped
In method org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]




--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4920) Stamped should implement equals

[Marc Juchli (JIRA)]

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

Marc Juchli updated KAFKA-4920:
---
Description: 
We ran FundBug, which resulted in the "Bad practice warning": 
```Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class org.apache.kafka.streams.processor.internals.Stamped
In method org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]```


  was:
We ran FundBug, which resulted in the "Bad practice warning": 
Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class org.apache.kafka.streams.processor.internals.Stamped
In method org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]



> Stamped should implement equals
> ---
>
> Key: KAFKA-4920
> URL: https://issues.apache.org/jira/browse/KAFKA-4920
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Marc Juchli
>Priority: Minor
>
> We ran FundBug, which resulted in the "Bad practice warning": 
> ```Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
> In class org.apache.kafka.streams.processor.internals.Stamped
> In method 
> org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
> At Stamped.java:[lines 31-35]```



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4920) Stamped should implement equals

[Marc Juchli (JIRA)]

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

Marc Juchli updated KAFKA-4920:
---
Description: 
We ran FundBug, which resulted in the "Bad practice warning": 

Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class org.apache.kafka.streams.processor.internals.Stamped
In method org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]


  was:
We ran FundBug, which resulted in the "Bad practice warning": 
`Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class org.apache.kafka.streams.processor.internals.Stamped
In method org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]`



> Stamped should implement equals
> ---
>
> Key: KAFKA-4920
> URL: https://issues.apache.org/jira/browse/KAFKA-4920
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Marc Juchli
>Priority: Minor
>
> We ran FundBug, which resulted in the "Bad practice warning": 
> Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
> In class org.apache.kafka.streams.processor.internals.Stamped
> In method 
> org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
> At Stamped.java:[lines 31-35]



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4920) Stamped should implement equals

[Marc Juchli (JIRA)]

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

Marc Juchli updated KAFKA-4920:
---
Description: 
We ran FundBug, which resulted in the "Bad practice warning": 
`Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class org.apache.kafka.streams.processor.internals.Stamped
In method org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]`


  was:
We ran FundBug, which resulted in the "Bad practice warning": 
```Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
In class org.apache.kafka.streams.processor.internals.Stamped
In method org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
At Stamped.java:[lines 31-35]```



> Stamped should implement equals
> ---
>
> Key: KAFKA-4920
> URL: https://issues.apache.org/jira/browse/KAFKA-4920
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Marc Juchli
>Priority: Minor
>
> We ran FundBug, which resulted in the "Bad practice warning": 
> `Bug type EQ_COMPARETO_USE_OBJECT_EQUALS (click for details) 
> In class org.apache.kafka.streams.processor.internals.Stamped
> In method 
> org.apache.kafka.streams.processor.internals.Stamped.compareTo(Object)
> At Stamped.java:[lines 31-35]`



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[GitHub] kafka pull request #2709: MINOR: Map `mkString` format updated to default ja...

[Kamal15]

GitHub user Kamal15 opened a pull request:

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

MINOR: Map `mkString` format updated to default java format

This is a minor change but it helps to improve the log readability.

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

$ git pull https://github.com/Kamal15/kafka util

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

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


commit 72b8e6f63ab7830dc04b9ff6da075eea8f990996
Author: Kamal C 
Date:   2017-03-20T12:54:44Z

MINOR: Map `mkString` format updated to default java format




---
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-4916) Add streams tests with brokers failing

[Eno Thereska (JIRA)]

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

Eno Thereska reassigned KAFKA-4916:
---

Assignee: Eno Thereska

> Add streams tests with brokers failing
> --
>
> Key: KAFKA-4916
> URL: https://issues.apache.org/jira/browse/KAFKA-4916
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.2.0
>Reporter: Eno Thereska
>Assignee: Eno Thereska
>Priority: Critical
> Fix For: 0.11.0.0
>
>
> We need to add either integration or system tests with streams and have Kafka 
> brokers fail and come back up. A combination of transient and permanent 
> broker failures.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)