[jira] [Assigned] (KAFKA-3556) Improve group coordinator metrics

[Liquan Pei (JIRA)]

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

Liquan Pei reassigned KAFKA-3556:
-

Assignee: Liquan Pei

> Improve group coordinator metrics
> -
>
> Key: KAFKA-3556
> URL: https://issues.apache.org/jira/browse/KAFKA-3556
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Jason Gustafson
>Assignee: Liquan Pei
>
> We currently don't have many metrics to track the behavior of the group 
> coordinator (especially with respect to the new consumer). On a quick pass, I 
> only saw a couple gauges in GroupMetadataManager for the number of groups and 
> the number of cached offsets. Here are some interesting metrics that may be 
> worth tracking:
> 1. Session timeout rate
> 2. Rebalance latency/rate
> 3. Commit latency/rate
> 4. Average group size
> 5. Size of metadata cache
> Some of these may also be interesting to track per group.



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

[jira] [Resolved] (KAFKA-2479) Add CopycatExceptions to indicate transient and permanent errors in a connector/task

[Liquan Pei (JIRA)]

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

Liquan Pei resolved KAFKA-2479.
---
Resolution: Fixed

> Add CopycatExceptions to indicate transient and permanent errors in a 
> connector/task
> 
>
> Key: KAFKA-2479
> URL: https://issues.apache.org/jira/browse/KAFKA-2479
> Project: Kafka
>  Issue Type: Sub-task
>  Components: KafkaConnect
>Reporter: Ewen Cheslack-Postava
>Assignee: Liquan Pei
> Fix For: 0.10.1.0
>
>
> Sometimes the connector will need to indicate to the framework that an error 
> occurred, but the error could have multiple responses by the framework.
> For source connectors, there's not much they need to indicate since they can 
> block indefinitely. They probably only need to indicate permanent errors for 
> correctness, though we may want them to indicate transient errors so we can 
> report health of the task in a metric.
> For sink connectors, there are at least a couple of scenarios:
> 1. A task encounters some error while processing a {{put(records)}} call and 
> was unable to fully process it, but thinks it could be resolved in the 
> future. The task doesn't want to see any new records until the issue is 
> resolved, but will need to see the same set of records again. (It would be 
> nice if the task doesn't have to deal with saving these to a buffer itself.)
> 2. A task encounters some error while processing data, but it has 
> enqueued/handled the data passed into the {{put(records)}} call. For example, 
> it may have passed it to some library which buffers it, but then the library 
> indicated that it is having some connection issues. The connector might be 
> able accept more data, but the task is not in a healthy state.
> 3. The task encounters some error that it decides is unrecoverable. This 
> might just be transient errors that repeat for long enough that the task 
> thinks its time to give up. Unclear what to do here, but one option is 
> relocating the task to another worker, hoping that the issue is specific to 
> the worker.
> Note that it is not, generally, safe for sink tasks to do their own backoff 
> or we'd potentially starve the consumer, which needs to poll() in order to 
> heartbeat. So we need to make sure whatever mechanism we implement encourages 
> the user to throw an exception and pass control back to us instead.



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

[jira] [Assigned] (KAFKA-3527) Consumer commitAsync should not expose internal exceptions

[Liquan Pei (JIRA)]

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

Liquan Pei reassigned KAFKA-3527:
-

Assignee: Liquan Pei  (was: Jason Gustafson)

> Consumer commitAsync should not expose internal exceptions
> --
>
> Key: KAFKA-3527
> URL: https://issues.apache.org/jira/browse/KAFKA-3527
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.0, 0.9.0.1
>Reporter: Jason Gustafson
>Assignee: Liquan Pei
>
> Currently we expose some internal exceptions to the user in the consumer's 
> OffsetCommitCallback (e.g. group load in progress, not coordinator for 
> group). We should convert these to instances of CommitFailedException and 
> provide a clear message to the user.



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

[jira] [Assigned] (KAFKA-3457) KafkaConsumer.committed(...) hangs forever if port number is wrong

[Liquan Pei (JIRA)]

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

Liquan Pei reassigned KAFKA-3457:
-

Assignee: Liquan Pei

> KafkaConsumer.committed(...) hangs forever if port number is wrong
> --
>
> Key: KAFKA-3457
> URL: https://issues.apache.org/jira/browse/KAFKA-3457
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.9.0.1
>Reporter: Harald Kirsch
>Assignee: Liquan Pei
>
> Create a KafkaConsumer with default settings but with a wrong host:port 
> setting for bootstrap.servers. Have it in some consumer group, do not 
> subscribe or assign partitions.
> Then call .committed(...) for a topic/partition combination a few times. It 
> will hang on the 2nd or third call forever. In the debug log you will see 
> that it repeats connections all over again. I waited many minutes and it 
> never came back to throw an Exception.
> The connections problems should at least pop out on the WARNING log level. 
> Likely the connection problems should throw an exception eventually.



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

[jira] [Reopened] (KAFKA-3605) Connector REST endpoint allows incorrectly overriding the connector name

[Liquan Pei (JIRA)]

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

Liquan Pei reopened KAFKA-3605:
---

> Connector REST endpoint allows incorrectly overriding the connector name
> 
>
> Key: KAFKA-3605
> URL: https://issues.apache.org/jira/browse/KAFKA-3605
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.9.0.1
>Reporter: Ewen Cheslack-Postava
>Assignee: Ewen Cheslack-Postava
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> If you PUT a connector config and it includes a connector name that doesn't 
> match the URL, the request will succeed. It obviously shouldn't since the 
> wrong connector name will be included. This should trigger an error in the 
> 400 range instead.



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

[jira] [Resolved] (KAFKA-3605) Connector REST endpoint allows incorrectly overriding the connector name

[Liquan Pei (JIRA)]

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

Liquan Pei resolved KAFKA-3605.
---
Resolution: Fixed

> Connector REST endpoint allows incorrectly overriding the connector name
> 
>
> Key: KAFKA-3605
> URL: https://issues.apache.org/jira/browse/KAFKA-3605
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.9.0.1
>Reporter: Ewen Cheslack-Postava
>Assignee: Ewen Cheslack-Postava
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> If you PUT a connector config and it includes a connector name that doesn't 
> match the URL, the request will succeed. It obviously shouldn't since the 
> wrong connector name will be included. This should trigger an error in the 
> 400 range instead.



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

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

[Apache Jenkins Server]

See 

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

[Apache Jenkins Server]

See 

Changes:

[me] KAFKA-3606: Traverse CLASSPATH during herder start

--
[...truncated 5326 lines...]
org.apache.kafka.streams.processor.TopologyBuilderTest > 
testTopicGroupsByStateStore PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithDuplicates PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkWithWrongParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkConnectedWithMultipleParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddProcessorWithWrongParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > testAddStateStore 
PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkConnectedWithParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithNonExistingProcessor PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testRegisterNonPersistentStore PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testLockStateDirectory PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testGetStore PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testClose PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testChangeLogOffsets PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testRegisterPersistentStore PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testNoTopic PASSED

org.apache.kafka.streams.processor.internals.MinTimestampTrackerTest > 
testTracking PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > 
testStorePartitions PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > testUpdateKTable 
PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > 
testUpdateNonPersistentStore PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > testUpdate PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > testProcessOrder 
PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > testPauseResume 
PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > 
testMaybePunctuate PASSED

org.apache.kafka.streams.processor.internals.PartitionGroupTest > 
testTimeTracking PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testStickiness PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testAssignWithStandby PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testAssignWithoutStandby PASSED

org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfoTest > 
testEncodeDecode PASSED

org.apache.kafka.streams.processor.internals.assignment.AssginmentInfoTest > 
testEncodeDecode PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingMultiplexingTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingStatefulTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingSimpleTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testTopologyMetadata PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingMultiplexByNameTopology PASSED

org.apache.kafka.streams.processor.internals.QuickUnionTest > testUnite PASSED

org.apache.kafka.streams.processor.internals.QuickUnionTest > testUniteMany 
PASSED

org.apache.kafka.streams.processor.internals.PunctuationQueueTest > 
testPunctuationInterval PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithStandbyReplicas PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithNewTasks PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithStates PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignBasic PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testOnAssignment PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testSubscription PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithInternalTopics PASSED

org.apache.kafka.streams.processor.internals.RecordCollectorTest > 
testSpecificPartition PASSED

org.apache.kafka.streams.processor.internals.RecordCollectorTest > 
testStreamPartitioner PASSED

org.apache.kafka.streams.processor.internals.RecordQueueTest > testTimeTracking 
PASSED

org.apache.kafka.streams.processor.internals.StreamThreadTest > testMaybeCommit 
PASSED

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

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-3594; After calling MemoryRecords.close() method, hasRoomFor()

--
[...truncated 717 lines...]

kafka.server.KafkaConfigTest > testFromPropsInvalid PASSED

kafka.server.KafkaConfigTest > testInvalidCompressionType PASSED

kafka.server.KafkaConfigTest > testAdvertiseHostNameDefault PASSED

kafka.server.KafkaConfigTest > testLogRetentionTimeMinutesProvided PASSED

kafka.server.KafkaConfigTest > testValidCompressionType PASSED

kafka.server.KafkaConfigTest > testUncleanElectionInvalid PASSED

kafka.server.KafkaConfigTest > testLogRetentionTimeBothMinutesAndMsProvided 
PASSED

kafka.server.KafkaConfigTest > testLogRollTimeMsProvided PASSED

kafka.server.KafkaConfigTest > testUncleanLeaderElectionDefault PASSED

kafka.server.KafkaConfigTest > testInvalidAdvertisedListenersProtocol PASSED

kafka.server.KafkaConfigTest > testUncleanElectionEnabled PASSED

kafka.server.KafkaConfigTest > testAdvertisePortDefault PASSED

kafka.server.KafkaConfigTest > testVersionConfiguration PASSED

kafka.server.KafkaConfigTest > testEqualAdvertisedListenersProtocol PASSED

kafka.server.SimpleFetchTest > testReadFromLog PASSED

kafka.server.DelayedOperationTest > testRequestPurge PASSED

kafka.server.DelayedOperationTest > testRequestExpiry PASSED

kafka.server.DelayedOperationTest > testRequestSatisfaction PASSED

kafka.server.LogRecoveryTest > testHWCheckpointNoFailuresMultipleLogSegments 
PASSED

kafka.server.LogRecoveryTest > testHWCheckpointWithFailuresMultipleLogSegments 
PASSED

kafka.server.LogRecoveryTest > testHWCheckpointNoFailuresSingleLogSegment PASSED

kafka.server.LogRecoveryTest > testHWCheckpointWithFailuresSingleLogSegment 
PASSED

kafka.server.ServerShutdownTest > testCleanShutdownAfterFailedStartup PASSED

kafka.server.ServerShutdownTest > testConsecutiveShutdown PASSED

kafka.server.ServerShutdownTest > testCleanShutdown PASSED

kafka.server.ServerShutdownTest > testCleanShutdownWithDeleteTopicEnabled PASSED

kafka.server.ServerStartupTest > testBrokerCreatesZKChroot PASSED

kafka.server.ServerStartupTest > testConflictBrokerRegistration PASSED

kafka.server.ServerStartupTest > testBrokerSelfAware PASSED

kafka.server.EdgeCaseRequestTest > testInvalidApiVersionRequest PASSED

kafka.server.EdgeCaseRequestTest > testMalformedHeaderRequest PASSED

kafka.server.EdgeCaseRequestTest > testProduceRequestWithNullClientId PASSED

kafka.server.EdgeCaseRequestTest > testInvalidApiKeyRequest PASSED

kafka.server.EdgeCaseRequestTest > testHeaderOnlyRequest PASSED

kafka.server.OffsetCommitTest > testUpdateOffsets PASSED

kafka.server.OffsetCommitTest > testLargeMetadataPayload PASSED

kafka.server.OffsetCommitTest > testCommitAndFetchOffsets PASSED

kafka.server.OffsetCommitTest > testOffsetExpiration PASSED

kafka.server.OffsetCommitTest > testNonExistingTopicOffsetCommit PASSED

kafka.server.ReplicaManagerTest > 
testFetchBeyondHighWatermarkNotAllowedForConsumer PASSED

kafka.server.ReplicaManagerTest > testHighWaterMarkDirectoryMapping PASSED

kafka.server.ReplicaManagerTest > testIllegalRequiredAcks PASSED

kafka.server.ReplicaManagerTest > testClearPurgatoryOnBecomingFollower PASSED

kafka.server.ReplicaManagerTest > testHighwaterMarkRelativeDirectoryMapping 
PASSED

kafka.server.IsrExpirationTest > testIsrExpirationForSlowFollowers PASSED

kafka.server.IsrExpirationTest > testIsrExpirationForStuckFollowers PASSED

kafka.server.IsrExpirationTest > testIsrExpirationIfNoFetchRequestMade PASSED

kafka.server.ClientQuotaManagerTest > testQuotaViolation PASSED

kafka.server.ClientQuotaManagerTest > testExpireQuotaSensors PASSED

kafka.server.ClientQuotaManagerTest > testExpireThrottleTimeSensor PASSED

kafka.server.ClientQuotaManagerTest > testQuotaParsing PASSED

kafka.server.MetadataCacheTest > 
getTopicMetadataWithNonSupportedSecurityProtocol PASSED

kafka.server.MetadataCacheTest > getTopicMetadataIsrNotAvailable PASSED

kafka.server.MetadataCacheTest > getTopicMetadata PASSED

kafka.server.MetadataCacheTest > getTopicMetadataReplicaNotAvailable PASSED

kafka.server.MetadataCacheTest > getTopicMetadataPartitionLeaderNotAvailable 
PASSED

kafka.server.MetadataCacheTest > getAliveBrokersShouldNotBeMutatedByUpdateCache 
PASSED

kafka.server.MetadataCacheTest > getTopicMetadataNonExistingTopics PASSED

kafka.server.ServerGenerateBrokerIdTest > testBrokerMetadataOnIdCollision PASSED

kafka.server.ServerGenerateBrokerIdTest > testAutoGenerateBrokerId PASSED

kafka.server.ServerGenerateBrokerIdTest > testMultipleLogDirsMetaProps PASSED

kafka.server.ServerGenerateBrokerIdTest > testDisableGeneratedBrokerId PASSED

kafka.server.ServerGenerateBrokerIdTest > testUserConfigAndGeneratedBrokerId 
PASSED

kafka.server.ServerGenerateBrokerIdTest > 
testConsistentBrokerIdFromUserConfigAndMetaProps PASSED

kafka.server.LogOffsetTest > testGetOffsetsBeforeEarliestTime PASSED

kafka.server.LogOffsetTest > 

[jira] [Commented] (KAFKA-3606) Traverse CLASSPATH during herder start to list connectors

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Traverse CLASSPATH during herder start to list connectors
> -
>
> Key: KAFKA-3606
> URL: https://issues.apache.org/jira/browse/KAFKA-3606
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.0.0
>Reporter: Liquan Pei
>Assignee: Liquan Pei
>Priority: Blocker
> Fix For: 0.10.1.0
>
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> List connectors API requires CLASSPATH traversal, which can takes up to 30s  
> to return available connectors. To work around this, we traverse the 
> CLASSPATH when staring herder and cache the result. Also we should guard 
> against concurrent CLASSPATH traversal. 



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

[jira] [Resolved] (KAFKA-3606) Traverse CLASSPATH during herder start to list connectors

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava resolved KAFKA-3606.
--
   Resolution: Fixed
Fix Version/s: (was: 0.10.0.0)
   0.10.1.0

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

> Traverse CLASSPATH during herder start to list connectors
> -
>
> Key: KAFKA-3606
> URL: https://issues.apache.org/jira/browse/KAFKA-3606
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.10.0.0
>Reporter: Liquan Pei
>Assignee: Liquan Pei
>Priority: Blocker
> Fix For: 0.10.1.0
>
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> List connectors API requires CLASSPATH traversal, which can takes up to 30s  
> to return available connectors. To work around this, we traverse the 
> CLASSPATH when staring herder and cache the result. Also we should guard 
> against concurrent CLASSPATH traversal. 



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

[GitHub] kafka pull request: KAFKA-3606: Traverse CLASSPATH during herder s...

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Created] (KAFKA-3607) Close state stores explicitly in unit tests upon completing

[Guozhang Wang (JIRA)]

Guozhang Wang created KAFKA-3607:


 Summary: Close state stores explicitly in unit tests upon 
completing
 Key: KAFKA-3607
 URL: https://issues.apache.org/jira/browse/KAFKA-3607
 Project: Kafka
  Issue Type: Bug
  Components: streams
Reporter: Guozhang Wang
Assignee: Guozhang Wang


In places like {{KStreamTestDriver}}, one or more state stores can be created 
and initialized, but they are not explicitly closed at the end of the test. We 
suspect that it may be the cause of the underlying RocksDB "pure virtual method 
called" exception.



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

Re: [VOTE] KIP-35: Retrieve protocol version - Round 2

[Tom Crayford]

+1 (non binding)

On Friday, 22 April 2016, Jason Gustafson  wrote:

> +1 (non-binding)
>
> On Thu, Apr 21, 2016 at 5:26 PM, Ewen Cheslack-Postava  >
> wrote:
>
> > +1
> >
> > On Thu, Apr 21, 2016 at 5:25 PM, Guozhang Wang  > wrote:
> >
> > > +1
> > >
> > > On Thu, Apr 21, 2016 at 5:07 PM, Gwen Shapira  > wrote:
> > >
> > > > +1
> > > >
> > > > On Thu, Apr 21, 2016 at 4:48 PM, Ashish Singh  >
> > > wrote:
> > > > > Hey Guys,
> > > > >
> > > > > I would like to re-initiate the voting process for *KIP-35:
> Retrieve
> > > > > protocol version*.
> > > > >
> > > > > KIP-35 can be accessed here
> > > > > <
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-35+-+Retrieving+protocol+version
> > > > >.
> > > > > Following are a couple of related PRs.
> > > > >
> > > > >1. KAFKA-3307: Add ApiVersion request/response and server side
> > > > handling.
> > > > >
> > > > >2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp
> to
> > > > check
> > > > >if the broker they are talking to supports required api
> versions.
> > > > >
> > > > >
> > > > > The vote will run for 72 hours and I would like to start it with my
> > +1
> > > > > (non-binding).
> > > > >
> > > > > --
> > > > >
> > > > > Regards,
> > > > > Ashish
> > > >
> > >
> > >
> > >
> > > --
> > > -- Guozhang
> > >
> >
> >
> >
> > --
> > Thanks,
> > Ewen
> >
>

Re: [VOTE] KIP-35: Retrieve protocol version - Round 2

[Jason Gustafson]

+1 (non-binding)

On Thu, Apr 21, 2016 at 5:26 PM, Ewen Cheslack-Postava 
wrote:

> +1
>
> On Thu, Apr 21, 2016 at 5:25 PM, Guozhang Wang  wrote:
>
> > +1
> >
> > On Thu, Apr 21, 2016 at 5:07 PM, Gwen Shapira  wrote:
> >
> > > +1
> > >
> > > On Thu, Apr 21, 2016 at 4:48 PM, Ashish Singh 
> > wrote:
> > > > Hey Guys,
> > > >
> > > > I would like to re-initiate the voting process for *KIP-35: Retrieve
> > > > protocol version*.
> > > >
> > > > KIP-35 can be accessed here
> > > > <
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-35+-+Retrieving+protocol+version
> > > >.
> > > > Following are a couple of related PRs.
> > > >
> > > >1. KAFKA-3307: Add ApiVersion request/response and server side
> > > handling.
> > > >
> > > >2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to
> > > check
> > > >if the broker they are talking to supports required api versions.
> > > >
> > > >
> > > > The vote will run for 72 hours and I would like to start it with my
> +1
> > > > (non-binding).
> > > >
> > > > --
> > > >
> > > > Regards,
> > > > Ashish
> > >
> >
> >
> >
> > --
> > -- Guozhang
> >
>
>
>
> --
> Thanks,
> Ewen
>

[jira] [Commented] (KAFKA-3602) Rename RecordAccumulator dequeFor() and ensure proper usage

[ASF GitHub Bot (JIRA)]

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

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

GitHub user hachikuji opened a pull request:

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

KAFKA-3602: rename RecordAccumulator dequeFor() and fix usage



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

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

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

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






> Rename RecordAccumulator dequeFor() and ensure proper usage
> ---
>
> Key: KAFKA-3602
> URL: https://issues.apache.org/jira/browse/KAFKA-3602
> Project: Kafka
>  Issue Type: Bug
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>
> In the investigation of KAFKA-3358, we found a case where the side effect of 
> creating record batches in the internal RecordAccumulator.dequeFor() 
> method caused unintended behavior. The bug in this case an implicit 
> assumption elsewhere in the code that all record batches correspond to 
> partitions that have actually been targeted for sends by the user. This 
> assumption was invalidated when a topic metadata request is sent with no 
> topics, which results in the metadata from all topics being returned. The end 
> result when that happened is that the client gets stuck trying to fetch 
> metadata for topics which were not even used.
> Although this particular problem will be fixed by changing the TopicMetadata 
> request in KIP-4, it probably could have been avoided by making the 
> side-effect of batch creation clear in the method name. For example, instead 
> of dequeFor(), we should use something like getOrCreateDeque(). It's more 
> verbose, but it makes the behavior clear. From a scan of the code, it looks 
> like there are a couple places where we do not expect the side-effect of 
> batch creation, so we should fix that too.



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

[GitHub] kafka pull request: KAFKA-3602: rename RecordAccumulator dequeFor(...

[hachikuji]

GitHub user hachikuji opened a pull request:

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

KAFKA-3602: rename RecordAccumulator dequeFor() and fix usage



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

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

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

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






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

Re: [VOTE] KIP-35: Retrieve protocol version - Round 2

[Ewen Cheslack-Postava]

+1

On Thu, Apr 21, 2016 at 5:25 PM, Guozhang Wang  wrote:

> +1
>
> On Thu, Apr 21, 2016 at 5:07 PM, Gwen Shapira  wrote:
>
> > +1
> >
> > On Thu, Apr 21, 2016 at 4:48 PM, Ashish Singh 
> wrote:
> > > Hey Guys,
> > >
> > > I would like to re-initiate the voting process for *KIP-35: Retrieve
> > > protocol version*.
> > >
> > > KIP-35 can be accessed here
> > > <
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-35+-+Retrieving+protocol+version
> > >.
> > > Following are a couple of related PRs.
> > >
> > >1. KAFKA-3307: Add ApiVersion request/response and server side
> > handling.
> > >
> > >2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to
> > check
> > >if the broker they are talking to supports required api versions.
> > >
> > >
> > > The vote will run for 72 hours and I would like to start it with my +1
> > > (non-binding).
> > >
> > > --
> > >
> > > Regards,
> > > Ashish
> >
>
>
>
> --
> -- Guozhang
>



-- 
Thanks,
Ewen

Re: [VOTE] KIP-35: Retrieve protocol version - Round 2

[Guozhang Wang]

+1

On Thu, Apr 21, 2016 at 5:07 PM, Gwen Shapira  wrote:

> +1
>
> On Thu, Apr 21, 2016 at 4:48 PM, Ashish Singh  wrote:
> > Hey Guys,
> >
> > I would like to re-initiate the voting process for *KIP-35: Retrieve
> > protocol version*.
> >
> > KIP-35 can be accessed here
> > <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-35+-+Retrieving+protocol+version
> >.
> > Following are a couple of related PRs.
> >
> >1. KAFKA-3307: Add ApiVersion request/response and server side
> handling.
> >
> >2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to
> check
> >if the broker they are talking to supports required api versions.
> >
> >
> > The vote will run for 72 hours and I would like to start it with my +1
> > (non-binding).
> >
> > --
> >
> > Regards,
> > Ashish
>



-- 
-- Guozhang

Re: [VOTE] KIP-35: Retrieve protocol version - Round 2

[Gwen Shapira]

+1

On Thu, Apr 21, 2016 at 4:48 PM, Ashish Singh  wrote:
> Hey Guys,
>
> I would like to re-initiate the voting process for *KIP-35: Retrieve
> protocol version*.
>
> KIP-35 can be accessed here
> .
> Following are a couple of related PRs.
>
>1. KAFKA-3307: Add ApiVersion request/response and server side handling.
>
>2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to check
>if the broker they are talking to supports required api versions.
>
>
> The vote will run for 72 hours and I would like to start it with my +1
> (non-binding).
>
> --
>
> Regards,
> Ashish

[jira] [Created] (KAFKA-3606) Traverse CLASSPATH during herder start to list connectors

[Liquan Pei (JIRA)]

Liquan Pei created KAFKA-3606:
-

 Summary: Traverse CLASSPATH during herder start to list connectors
 Key: KAFKA-3606
 URL: https://issues.apache.org/jira/browse/KAFKA-3606
 Project: Kafka
  Issue Type: Bug
  Components: KafkaConnect
Affects Versions: 0.10.0.0
Reporter: Liquan Pei
Assignee: Liquan Pei
Priority: Blocker
 Fix For: 0.10.0.0


List connectors API requires CLASSPATH traversal, which can takes up to 30s  to 
return available connectors. To work around this, we traverse the CLASSPATH 
when staring herder and cache the result. Also we should guard against 
concurrent CLASSPATH traversal. 



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

[jira] [Commented] (KAFKA-3605) Connector REST endpoint allows incorrectly overriding the connector name

[ASF GitHub Bot (JIRA)]

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

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

GitHub user ewencp opened a pull request:

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

KAFKA-3605: Return error if connector config includes mismatching connector 
name.



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

$ git pull https://github.com/ewencp/kafka 
kafka-3605-connector-name-mismatch

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

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


commit 1ed45ec3e93252c15e905de9cfc14630303711b8
Author: Ewen Cheslack-Postava 
Date:   2016-04-21T23:48:37Z

KAFKA-3605: Return error if connector config includes mismatching connector 
name.




> Connector REST endpoint allows incorrectly overriding the connector name
> 
>
> Key: KAFKA-3605
> URL: https://issues.apache.org/jira/browse/KAFKA-3605
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 0.9.0.1
>Reporter: Ewen Cheslack-Postava
>Assignee: Ewen Cheslack-Postava
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> If you PUT a connector config and it includes a connector name that doesn't 
> match the URL, the request will succeed. It obviously shouldn't since the 
> wrong connector name will be included. This should trigger an error in the 
> 400 range instead.



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

[GitHub] kafka pull request: KAFKA-3605: Return error if connector config i...

[ewencp]

GitHub user ewencp opened a pull request:

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

KAFKA-3605: Return error if connector config includes mismatching connector 
name.



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

$ git pull https://github.com/ewencp/kafka 
kafka-3605-connector-name-mismatch

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

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


commit 1ed45ec3e93252c15e905de9cfc14630303711b8
Author: Ewen Cheslack-Postava 
Date:   2016-04-21T23:48:37Z

KAFKA-3605: Return error if connector config includes mismatching connector 
name.




---
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-35: Retrieve protocol version - Round 2

[Ashish Singh]

Hey Guys,

I would like to re-initiate the voting process for *KIP-35: Retrieve
protocol version*.

KIP-35 can be accessed here
.
Following are a couple of related PRs.

   1. KAFKA-3307: Add ApiVersion request/response and server side handling.
   
   2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to check
   if the broker they are talking to supports required api versions.
   

The vote will run for 72 hours and I would like to start it with my +1
(non-binding).
​
-- 

Regards,
Ashish

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Ashish Singh]

Done. Thanks!

On Thu, Apr 21, 2016 at 4:32 PM, Gwen Shapira  wrote:

> Lets start a vote immediately? We are short of time toward the release.
>
> On Thu, Apr 21, 2016 at 2:57 PM, Ashish Singh  wrote:
> > Hey Guys,
> >
> > KIP-35
> > <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-35+-+Retrieving+protocol+version
> >
> > has been updated based on latest discussions and following PRs have also
> > been updated.
> > 1. KAFKA-3307: Add ApiVersion request/response and server side handling.
> > 
> > 2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to check
> if
> > the broker they are talking to supports required api versions.
> > 
> >
> > If there are no major objections or changes suggested, we can start a
> vote
> > thread in a couple of days.
> >
> > On Tue, Apr 12, 2016 at 8:04 AM, Jun Rao  wrote:
> >
> >> Hi, Ismael,
> >>
> >> The SASL engine that we used is the SASL library, right? How did the C
> >> client generate those SASL tokens? Once a SASL mechanism is chosen, the
> >> subsequent tokens are determined, right? So, my feeling is that those
> >> tokens are part of SaslHandshakeRequest and are just extended across
> >> multiple network packets. So modeling those as independent requests
> feels
> >> weird. When documentation them, we really need to document those as a
> >> sequence, not individual isolated requests that can be issued
> >> in arbitrary order. The version id will only add confusion since we
> can't
> >> version the tokens independently. We could explicitly add the client id
> and
> >> correlation id in the header, but I am not sure if it's worth the
> >> additional complexity.
> >>
> >> Thanks,
> >>
> >> Jun
> >>
> >> On Tue, Apr 12, 2016 at 1:18 AM, Ismael Juma  wrote:
> >>
> >> > Hi Jun,
> >> >
> >> > I understand the point about the SASL tokens being similar to the SSL
> >> > handshake in a way. However, is there any SASL library that handles
> the
> >> > network communication for these tokens? I couldn't find any and
> without
> >> > that, there isn't much benefit in deviating from Kafka's protocol (we
> >> > basically save the space taken by the request header). It's worth
> >> > mentioning that we are already adding the message size before the
> opaque
> >> > bytes provided by the library, so one could say we are already
> extending
> >> > the protocol.
> >> >
> >> > If we leave versioning aside, adding the standard Kafka request
> header to
> >> > those messages may also help from a debugging perspective as would
> then
> >> > include client id and correlation id along with the message.
> >> >
> >> > Ismael
> >> >
> >> > On Tue, Apr 12, 2016 at 2:13 AM, Jun Rao  wrote:
> >> >
> >> > > Magnus,
> >> > >
> >> > > That sounds reasonable. To reduce the changes on the server side,
> I'd
> >> > > suggest the following minor tweaks on the proposal.
> >> > >
> >> > > 1. Continue supporting the separate SASL and SASL_SSL port.
> >> > >
> >> > > On SASL port, we support the new sequence
> >> > > ApiVersionRequest (optional), SaslHandshakeRequest, SASL tokens,
> >> > > regular
> >> > > requests
> >> > >
> >> > > On SASL_SSL port, we support the new sequence
> >> > > SSL handshake bytes, ApiVersionRequest (optional),
> >> > > SaslHandshakeRequest,
> >> > > SASL tokens, regular requests
> >> > >
> >> > > 2. We don't wrap SASL tokens in Kafka protocol. Similar to your
> >> argument
> >> > > about SSL handshake, those SASL tokens are generated by SASL library
> >> and
> >> > > Kafka doesn't really control its versioning. Kafka only controls the
> >> > > selection of SASL mechanism, which will be versioned in
> >> > > SaslHandshakeRequest.
> >> > >
> >> > > Does that work for you?
> >> > >
> >> > > Thanks,
> >> > >
> >> > > Jun
> >> > >
> >> > >
> >> > > On Mon, Apr 11, 2016 at 11:15 AM, Magnus Edenhill <
> mag...@edenhill.se>
> >> > > wrote:
> >> > >
> >> > > > Hey Jun, see inline
> >> > > >
> >> > > > 2016-04-11 19:19 GMT+02:00 Jun Rao :
> >> > > >
> >> > > > > Hi, Magnus,
> >> > > > >
> >> > > > > Let me understand your proposal in more details just from the
> >> > client's
> >> > > > > perspective. My understanding of your proposal is the following.
> >> > > > >
> >> > > > > On plaintext port, the client will send the following bytes in
> >> order.
> >> > > > > ApiVersionRequest, SaslHandshakeRequest, SASL tokens (if
> SASL
> >> is
> >> > > > > enabled), regular requests
> >> > > > >
> >> > > > > On SSL port, the client will send the following bytes in order.
> >> > > > > SSL handshake bytes, ApiVersionRequest,
> SaslHandshakeRequest,
> >> > SASL
> >> > > > > tokens (if SASL is enabled), regular requests
> >> > > > >
> >> > > >
> >> > > >
> >> > > > Yup!
> >> > > > "SASL tokens" is a series of proper Kafka protocol
> >> > SaslHandshakeRequests
> >> > > > 

[GitHub] kafka pull request: Traverse CLASSPATH during herder start

[Ishiihara]

GitHub user Ishiihara opened a pull request:

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

Traverse CLASSPATH during herder start

@ewencp Can you take a quick look?

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

$ git pull https://github.com/Ishiihara/kafka pre-list-connectors

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

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


commit 4b4a425a8f4042f04ca0f83142aff31cdea882d7
Author: Liquan Pei 
Date:   2016-04-21T23:42:34Z

Traverse CLASSPATH during herder start




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

[jira] [Created] (KAFKA-3605) Connector REST endpoint allows incorrectly overriding the connector name

[Ewen Cheslack-Postava (JIRA)]

Ewen Cheslack-Postava created KAFKA-3605:


 Summary: Connector REST endpoint allows incorrectly overriding the 
connector name
 Key: KAFKA-3605
 URL: https://issues.apache.org/jira/browse/KAFKA-3605
 Project: Kafka
  Issue Type: Bug
  Components: KafkaConnect
Affects Versions: 0.9.0.1
Reporter: Ewen Cheslack-Postava
Assignee: Ewen Cheslack-Postava
Priority: Blocker
 Fix For: 0.10.0.0


If you PUT a connector config and it includes a connector name that doesn't 
match the URL, the request will succeed. It obviously shouldn't since the wrong 
connector name will be included. This should trigger an error in the 400 range 
instead.



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

Re: [DISCUSS] KIP-50 - Enhance Authorizer interface to be aware of supported Principal Types

[Gwen Shapira]

I would like to suggest taking the discussion of "how to break Kafka
down into modules" outside the scope of KIP-50 and outside the scope
of the next release.

I understand that the current location of the authorizer API is not
ideal, but I want to point out that the scope was already expanded
from a new method to a complete rewrite of the authorizer. Is the
current location really bad enough to expand the scope into larger
refactoring of Kafka?

Gwen

On Wed, Apr 20, 2016 at 10:43 PM, Ismael Juma  wrote:
> Hi Jay,
>
> Thanks for summarising the reasoning for the current approach. On the topic
> of additional jars, the obvious example that came up recently is sharing
> JSON serializers between connect and streams. Given the desire not to add a
> Jackson dependency to clients, it seems like adding a kafka-serializer-json
> (or something like that) may be needed. This is similar to the
> kafka-log4j-appender jar that we have today.
>
> When you look at it this way, then the situation is not as clear-cut as
> initially described. Perhaps a way to explain this is that we only add
> additional modules when they introduce a new dependency.
>
> Finally, it seems a bit weird to add something to `common` that is, in
> fact, not common. Would it not make sense to have a separate package for
> pluggable core/server classes (because they are pluggable we want them to
> be in Java and not to be associated with a particular Scala version)?
>
> Ismael
>
> On Wed, Apr 20, 2016 at 4:52 PM, Jay Kreps  wrote:
>
>> Yeah our take when we came up with this approach was pretty much what Gwen
>> is saying:
>> 1. In practice you either need the server or client to do anything and the
>> server depends on the client so bundling common and client doesn't hurt.
>> 2. Our experience with more granular jars (not in Kafka) was that although
>> it feels "cleaner" the complexity comes quickly for a few reasons. First it
>> gets hard to detangle the more granular packages (e.g. somebody needs to
>> use something in Utils in the authorizer package and then you no longer
>> have a dag). Second people end up mixing and matching in ways you didn't
>> anticipate which causes crazy heisenbugs (e.g. they depend on two different
>> versions of the client via transitive dependencies and somehow end up with
>> client version x and common version y due to duplicate entries on the class
>> path).
>>
>> I'm not really arguing that this approach is superior, I'm just saying this
>> is the current approach and that is the reason we went with it.
>>
>> So I could see splitting common and client and you could even further split
>> the producer and consumer and multiple sub-jars in common, and if this was
>> the approach I think a separate authorizer jar would make sense. But in the
>> current approach I think the authorizer stuff would be most consistent as a
>> public package in common. It is true that this means you build against more
>> stuff then needed but I'm not sure this has any negative implications in
>> practice.
>>
>> -Jay
>>
>> On Wed, Apr 20, 2016 at 4:17 PM, Gwen Shapira  wrote:
>>
>> > But its just a compile-time dependency, right?
>> > Since the third-party-authorizer-implementation will be installed on a
>> > broker where the common classes will exist anyway.
>> >
>> >
>> > On Wed, Apr 20, 2016 at 3:13 PM, Ashish Singh 
>> wrote:
>> > > Jay,
>> > >
>> > > Thanks for the info. I think having common in clients jar makes sense,
>> as
>> > > their is no direct usage of it. i.e., without depending on or using
>> > > clients. Authorizer is a bit different, as third party implementations
>> do
>> > > not really need anything from clients or server, all they need is
>> > > Authorizer interface and related classes. If we move authorizer into
>> > > common, then third party implementations will have to depend on
>> clients.
>> > > Though third party implementations depending on clients is not a big
>> > > problem, right now they depend on core, I think it is cleaner to have
>> > > dependency on minimal modules. Would you agree?
>> > >
>> > > On Wed, Apr 20, 2016 at 2:27 PM, Jay Kreps  wrote:
>> > >
>> > >> I think it's great that we're moving the interface to java and fixing
>> > some
>> > >> of the naming foibles.
>> > >>
>> > >> This isn't explicit in the KIP which just refers to the java package
>> > name
>> > >> (I think), but it looks like you are proposing adding a new authorizer
>> > jar
>> > >> for this new package and adding it as a dependency for the client jar.
>> > This
>> > >> is a bit inconsistent with how we decided to package stuff when we
>> > started
>> > >> with the new clients so it'd be good to work that out.
>> > >>
>> > >> To date the categorization has been:
>> > >> 1. Anything which is just in the clients is in org.apache.clients
>> under
>> > >> clients/
>> > >> 2. Anything which is in the server is kafka.* which is under 

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Gwen Shapira]

Lets start a vote immediately? We are short of time toward the release.

On Thu, Apr 21, 2016 at 2:57 PM, Ashish Singh  wrote:
> Hey Guys,
>
> KIP-35
> 
> has been updated based on latest discussions and following PRs have also
> been updated.
> 1. KAFKA-3307: Add ApiVersion request/response and server side handling.
> 
> 2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to check if
> the broker they are talking to supports required api versions.
> 
>
> If there are no major objections or changes suggested, we can start a vote
> thread in a couple of days.
>
> On Tue, Apr 12, 2016 at 8:04 AM, Jun Rao  wrote:
>
>> Hi, Ismael,
>>
>> The SASL engine that we used is the SASL library, right? How did the C
>> client generate those SASL tokens? Once a SASL mechanism is chosen, the
>> subsequent tokens are determined, right? So, my feeling is that those
>> tokens are part of SaslHandshakeRequest and are just extended across
>> multiple network packets. So modeling those as independent requests feels
>> weird. When documentation them, we really need to document those as a
>> sequence, not individual isolated requests that can be issued
>> in arbitrary order. The version id will only add confusion since we can't
>> version the tokens independently. We could explicitly add the client id and
>> correlation id in the header, but I am not sure if it's worth the
>> additional complexity.
>>
>> Thanks,
>>
>> Jun
>>
>> On Tue, Apr 12, 2016 at 1:18 AM, Ismael Juma  wrote:
>>
>> > Hi Jun,
>> >
>> > I understand the point about the SASL tokens being similar to the SSL
>> > handshake in a way. However, is there any SASL library that handles the
>> > network communication for these tokens? I couldn't find any and without
>> > that, there isn't much benefit in deviating from Kafka's protocol (we
>> > basically save the space taken by the request header). It's worth
>> > mentioning that we are already adding the message size before the opaque
>> > bytes provided by the library, so one could say we are already extending
>> > the protocol.
>> >
>> > If we leave versioning aside, adding the standard Kafka request header to
>> > those messages may also help from a debugging perspective as would then
>> > include client id and correlation id along with the message.
>> >
>> > Ismael
>> >
>> > On Tue, Apr 12, 2016 at 2:13 AM, Jun Rao  wrote:
>> >
>> > > Magnus,
>> > >
>> > > That sounds reasonable. To reduce the changes on the server side, I'd
>> > > suggest the following minor tweaks on the proposal.
>> > >
>> > > 1. Continue supporting the separate SASL and SASL_SSL port.
>> > >
>> > > On SASL port, we support the new sequence
>> > > ApiVersionRequest (optional), SaslHandshakeRequest, SASL tokens,
>> > > regular
>> > > requests
>> > >
>> > > On SASL_SSL port, we support the new sequence
>> > > SSL handshake bytes, ApiVersionRequest (optional),
>> > > SaslHandshakeRequest,
>> > > SASL tokens, regular requests
>> > >
>> > > 2. We don't wrap SASL tokens in Kafka protocol. Similar to your
>> argument
>> > > about SSL handshake, those SASL tokens are generated by SASL library
>> and
>> > > Kafka doesn't really control its versioning. Kafka only controls the
>> > > selection of SASL mechanism, which will be versioned in
>> > > SaslHandshakeRequest.
>> > >
>> > > Does that work for you?
>> > >
>> > > Thanks,
>> > >
>> > > Jun
>> > >
>> > >
>> > > On Mon, Apr 11, 2016 at 11:15 AM, Magnus Edenhill 
>> > > wrote:
>> > >
>> > > > Hey Jun, see inline
>> > > >
>> > > > 2016-04-11 19:19 GMT+02:00 Jun Rao :
>> > > >
>> > > > > Hi, Magnus,
>> > > > >
>> > > > > Let me understand your proposal in more details just from the
>> > client's
>> > > > > perspective. My understanding of your proposal is the following.
>> > > > >
>> > > > > On plaintext port, the client will send the following bytes in
>> order.
>> > > > > ApiVersionRequest, SaslHandshakeRequest, SASL tokens (if SASL
>> is
>> > > > > enabled), regular requests
>> > > > >
>> > > > > On SSL port, the client will send the following bytes in order.
>> > > > > SSL handshake bytes, ApiVersionRequest, SaslHandshakeRequest,
>> > SASL
>> > > > > tokens (if SASL is enabled), regular requests
>> > > > >
>> > > >
>> > > >
>> > > > Yup!
>> > > > "SASL tokens" is a series of proper Kafka protocol
>> > SaslHandshakeRequests
>> > > > until
>> > > > the handshake is done.
>> > > >
>> > > >
>> > > >
>> > > > >
>> > > > > Is that right? Since we can use either SSL or SASL for
>> > authentication,
>> > > > it's
>> > > > > weird that in one case, we require ApiVersionRequest to happen
>> before
>> > > > > authentication and in another case we require the reverse.
>> > > > >
>> > > >
>> > > > Since 

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

[Apache Jenkins Server]

See 

Changes:

[wangguoz] KAFKA-3337: Extract selector as a separate groupBy operator for 
KTable

[wangguoz] KAFKA-3589: set inner serializer for ChangedSerde upon initialization

--
[...truncated 5281 lines...]
org.apache.kafka.streams.processor.TopologyBuilderTest > 
testTopicGroupsByStateStore PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithDuplicates PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkWithWrongParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkConnectedWithMultipleParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddProcessorWithWrongParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > testAddStateStore 
PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkConnectedWithParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithNonExistingProcessor PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testRegisterNonPersistentStore PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testLockStateDirectory PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testGetStore PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testClose PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testChangeLogOffsets PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testRegisterPersistentStore PASSED

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testNoTopic PASSED

org.apache.kafka.streams.processor.internals.MinTimestampTrackerTest > 
testTracking PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > 
testStorePartitions PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > testUpdateKTable 
PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > 
testUpdateNonPersistentStore PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > testUpdate PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > testProcessOrder 
PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > testPauseResume 
PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > 
testMaybePunctuate PASSED

org.apache.kafka.streams.processor.internals.PartitionGroupTest > 
testTimeTracking PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testStickiness PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testAssignWithStandby PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testAssignWithoutStandby PASSED

org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfoTest > 
testEncodeDecode PASSED

org.apache.kafka.streams.processor.internals.assignment.AssginmentInfoTest > 
testEncodeDecode PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingMultiplexingTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingStatefulTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingSimpleTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testTopologyMetadata PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingMultiplexByNameTopology PASSED

org.apache.kafka.streams.processor.internals.QuickUnionTest > testUnite PASSED

org.apache.kafka.streams.processor.internals.QuickUnionTest > testUniteMany 
PASSED

org.apache.kafka.streams.processor.internals.PunctuationQueueTest > 
testPunctuationInterval PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithStandbyReplicas PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithNewTasks PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithStates PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignBasic PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testOnAssignment PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testSubscription PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithInternalTopics PASSED

org.apache.kafka.streams.processor.internals.RecordCollectorTest > 
testSpecificPartition PASSED

org.apache.kafka.streams.processor.internals.RecordCollectorTest > 
testStreamPartitioner PASSED

org.apache.kafka.streams.processor.internals.RecordQueueTest > testTimeTracking 
PASSED

Re: Kafka Streams - read topic from beginning

[Guozhang Wang]

Hello Maria,

We have some thoughts about supporting finer grained flow controls in Kafka
Streams https://issues.apache.org/jira/browse/KAFKA-3478 as part of a big
effort to improve re-processing user experience, which covers this use
case. We are shooting to have this post 0.10.0.0.

As for now, one work-around I can think of is that upon restart /
re-processing, you can delete the offsets through an admin request (look at
ConsumerGroupCommand).


Guozhang


On Thu, Apr 21, 2016 at 6:19 AM, Maria Abramiuc 
wrote:

> Kafka Streams look great, but there is one thing I don't seem to find a way
> to do:
>
> - read a topic from beginning even if there is a offset saved:
>
>  I have :
>
>  props.put(StreamsConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
>
> this works as described if there is no offset save.
>
>   For a normal consumer we have:
>
>   seekToBeginning
>
>
> In KafkaConsumer:
>
> Line 132:
>
> NetworkClient netClient = new NetworkClient(new
> Selector(config.getLong("connections.max.idle.ms").longValue(),
> this.metrics, this.time, metricGrpPrefix, metricsTags,
> channelBuilder), this.metadata, this.clientId, 100,
> config.getLong("reconnect.backoff.ms").longValue(),
> config.getInt("send.buffer.bytes").intValue(),
> config.getInt("receive.buffer.bytes").intValue(),
> config.getInt("request.timeout.ms").intValue(), this.time);
> this.client = new ConsumerNetworkClient(netClient, this.metadata,
> this.time, this.retryBackoffMs);
> OffsetResetStrategy offsetResetStrategy =
>
> OffsetResetStrategy.valueOf(config.getString("auto.offset.reset").toUpperCase());
> this.subscriptions = new SubscriptionState(offsetResetStrategy);
>
> I can't find any way to set the consumer using StreamsConfig
> properties to seekToBeginning or to set subscriptionState to need
> offset reset.
>
>
>
> *Is there a way to force the consumption of a topic from begging using
> Kafka Streams?*
>
>
> Thank you for all the support provided,
>
> Maria Abramiuc
>



-- 
-- Guozhang

[jira] [Commented] (KAFKA-3144) report members with no assigned partitions in ConsumerGroupCommand

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian commented on KAFKA-3144:


[~Konrad Kalita], are you still working on this JIRA? If not, I'd be happy to 
take over. Thanks.

> report members with no assigned partitions in ConsumerGroupCommand
> --
>
> Key: KAFKA-3144
> URL: https://issues.apache.org/jira/browse/KAFKA-3144
> Project: Kafka
>  Issue Type: Improvement
>Affects Versions: 0.9.0.0
>Reporter: Jun Rao
>Assignee: Konrad Kalita
>  Labels: newbie
>
> A couple of suggestions on improving ConsumerGroupCommand. 
> 1. It would be useful to list members with no assigned partitions when doing 
> describe in ConsumerGroupCommand.
> 2. Currently, we show the client.id of each member when doing describe in 
> ConsumerGroupCommand. Since client.id is supposed to be the logical 
> application id, all members in the same group are supposed to set the same 
> client.id. So, it would be clearer if we show the client id as well as the 
> member id.



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

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

[Apache Jenkins Server]

See 

[jira] [Updated] (KAFKA-3594) Kafka new producer retries doesn't work in 0.9.0.1

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3594:
-
Resolution: Fixed
Status: Resolved  (was: Patch Available)

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

> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Promise$class.complete(Promise.scala:55) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:67)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:82)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch.run(BatchingExecutor.scala:58) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> 

[jira] [Commented] (KAFKA-3594) Kafka new producer retries doesn't work in 0.9.0.1

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Promise$class.complete(Promise.scala:55) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:67)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:82)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch.run(BatchingExecutor.scala:58) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
>  

[GitHub] kafka pull request: KAFKA-3594; After calling MemoryRecords.close(...

[asfgit]

Github user asfgit closed the pull request at:

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


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

[jira] [Updated] (KAFKA-3602) Rename RecordAccumulator dequeFor() and ensure proper usage

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3602:
-
Summary: Rename RecordAccumulator dequeFor() and ensure proper usage  (was: 
Rename RecordAccumulator dequeueFor() and ensure proper usage)

> Rename RecordAccumulator dequeFor() and ensure proper usage
> ---
>
> Key: KAFKA-3602
> URL: https://issues.apache.org/jira/browse/KAFKA-3602
> Project: Kafka
>  Issue Type: Bug
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>
> In the investigation of KAFKA-3358, we found a case where the side effect of 
> creating record batches in the internal RecordAccumulator.dequeueFor() 
> method caused unintended behavior. The bug in this case an implicit 
> assumption elsewhere in the code that all record batches correspond to 
> partitions that have actually been targeted for sends by the user. This 
> assumption was invalidated when a topic metadata request is sent with no 
> topics, which results in the metadata from all topics being returned. The end 
> result when that happened is that the client gets stuck trying to fetch 
> metadata for topics which were not even used.
> Although this particular problem will be fixed by changing the TopicMetadata 
> request in KIP-4, it probably could have been avoided by making the 
> side-effect of batch creation clear in the method name. For example, instead 
> of dequeFor(), we should use something like getOrCreateDeque(). It's more 
> verbose, but it makes the behavior clear. From a scan of the code, it looks 
> like there are a couple places where we do not expect the side-effect of 
> batch creation, so we should fix that too.



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

[jira] [Updated] (KAFKA-3602) Rename RecordAccumulator dequeFor() and ensure proper usage

[Guozhang Wang (JIRA)]

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

Guozhang Wang updated KAFKA-3602:
-
Description: 
In the investigation of KAFKA-3358, we found a case where the side effect of 
creating record batches in the internal RecordAccumulator.dequeFor() 
method caused unintended behavior. The bug in this case an implicit assumption 
elsewhere in the code that all record batches correspond to partitions that 
have actually been targeted for sends by the user. This assumption was 
invalidated when a topic metadata request is sent with no topics, which results 
in the metadata from all topics being returned. The end result when that 
happened is that the client gets stuck trying to fetch metadata for topics 
which were not even used.

Although this particular problem will be fixed by changing the TopicMetadata 
request in KIP-4, it probably could have been avoided by making the side-effect 
of batch creation clear in the method name. For example, instead of dequeFor(), 
we should use something like getOrCreateDeque(). It's more verbose, but it 
makes the behavior clear. From a scan of the code, it looks like there are a 
couple places where we do not expect the side-effect of batch creation, so we 
should fix that too.

  was:
In the investigation of KAFKA-3358, we found a case where the side effect of 
creating record batches in the internal RecordAccumulator.dequeueFor() 
method caused unintended behavior. The bug in this case an implicit assumption 
elsewhere in the code that all record batches correspond to partitions that 
have actually been targeted for sends by the user. This assumption was 
invalidated when a topic metadata request is sent with no topics, which results 
in the metadata from all topics being returned. The end result when that 
happened is that the client gets stuck trying to fetch metadata for topics 
which were not even used.

Although this particular problem will be fixed by changing the TopicMetadata 
request in KIP-4, it probably could have been avoided by making the side-effect 
of batch creation clear in the method name. For example, instead of dequeFor(), 
we should use something like getOrCreateDeque(). It's more verbose, but it 
makes the behavior clear. From a scan of the code, it looks like there are a 
couple places where we do not expect the side-effect of batch creation, so we 
should fix that too.


> Rename RecordAccumulator dequeFor() and ensure proper usage
> ---
>
> Key: KAFKA-3602
> URL: https://issues.apache.org/jira/browse/KAFKA-3602
> Project: Kafka
>  Issue Type: Bug
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>
> In the investigation of KAFKA-3358, we found a case where the side effect of 
> creating record batches in the internal RecordAccumulator.dequeFor() 
> method caused unintended behavior. The bug in this case an implicit 
> assumption elsewhere in the code that all record batches correspond to 
> partitions that have actually been targeted for sends by the user. This 
> assumption was invalidated when a topic metadata request is sent with no 
> topics, which results in the metadata from all topics being returned. The end 
> result when that happened is that the client gets stuck trying to fetch 
> metadata for topics which were not even used.
> Although this particular problem will be fixed by changing the TopicMetadata 
> request in KIP-4, it probably could have been avoided by making the 
> side-effect of batch creation clear in the method name. For example, instead 
> of dequeFor(), we should use something like getOrCreateDeque(). It's more 
> verbose, but it makes the behavior clear. From a scan of the code, it looks 
> like there are a couple places where we do not expect the side-effect of 
> batch creation, so we should fix that too.



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

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Ashish Singh]

Hey Guys,

KIP-35

has been updated based on latest discussions and following PRs have also
been updated.
1. KAFKA-3307: Add ApiVersion request/response and server side handling.

2. KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to check if
the broker they are talking to supports required api versions.


If there are no major objections or changes suggested, we can start a vote
thread in a couple of days.

On Tue, Apr 12, 2016 at 8:04 AM, Jun Rao  wrote:

> Hi, Ismael,
>
> The SASL engine that we used is the SASL library, right? How did the C
> client generate those SASL tokens? Once a SASL mechanism is chosen, the
> subsequent tokens are determined, right? So, my feeling is that those
> tokens are part of SaslHandshakeRequest and are just extended across
> multiple network packets. So modeling those as independent requests feels
> weird. When documentation them, we really need to document those as a
> sequence, not individual isolated requests that can be issued
> in arbitrary order. The version id will only add confusion since we can't
> version the tokens independently. We could explicitly add the client id and
> correlation id in the header, but I am not sure if it's worth the
> additional complexity.
>
> Thanks,
>
> Jun
>
> On Tue, Apr 12, 2016 at 1:18 AM, Ismael Juma  wrote:
>
> > Hi Jun,
> >
> > I understand the point about the SASL tokens being similar to the SSL
> > handshake in a way. However, is there any SASL library that handles the
> > network communication for these tokens? I couldn't find any and without
> > that, there isn't much benefit in deviating from Kafka's protocol (we
> > basically save the space taken by the request header). It's worth
> > mentioning that we are already adding the message size before the opaque
> > bytes provided by the library, so one could say we are already extending
> > the protocol.
> >
> > If we leave versioning aside, adding the standard Kafka request header to
> > those messages may also help from a debugging perspective as would then
> > include client id and correlation id along with the message.
> >
> > Ismael
> >
> > On Tue, Apr 12, 2016 at 2:13 AM, Jun Rao  wrote:
> >
> > > Magnus,
> > >
> > > That sounds reasonable. To reduce the changes on the server side, I'd
> > > suggest the following minor tweaks on the proposal.
> > >
> > > 1. Continue supporting the separate SASL and SASL_SSL port.
> > >
> > > On SASL port, we support the new sequence
> > > ApiVersionRequest (optional), SaslHandshakeRequest, SASL tokens,
> > > regular
> > > requests
> > >
> > > On SASL_SSL port, we support the new sequence
> > > SSL handshake bytes, ApiVersionRequest (optional),
> > > SaslHandshakeRequest,
> > > SASL tokens, regular requests
> > >
> > > 2. We don't wrap SASL tokens in Kafka protocol. Similar to your
> argument
> > > about SSL handshake, those SASL tokens are generated by SASL library
> and
> > > Kafka doesn't really control its versioning. Kafka only controls the
> > > selection of SASL mechanism, which will be versioned in
> > > SaslHandshakeRequest.
> > >
> > > Does that work for you?
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > >
> > > On Mon, Apr 11, 2016 at 11:15 AM, Magnus Edenhill 
> > > wrote:
> > >
> > > > Hey Jun, see inline
> > > >
> > > > 2016-04-11 19:19 GMT+02:00 Jun Rao :
> > > >
> > > > > Hi, Magnus,
> > > > >
> > > > > Let me understand your proposal in more details just from the
> > client's
> > > > > perspective. My understanding of your proposal is the following.
> > > > >
> > > > > On plaintext port, the client will send the following bytes in
> order.
> > > > > ApiVersionRequest, SaslHandshakeRequest, SASL tokens (if SASL
> is
> > > > > enabled), regular requests
> > > > >
> > > > > On SSL port, the client will send the following bytes in order.
> > > > > SSL handshake bytes, ApiVersionRequest, SaslHandshakeRequest,
> > SASL
> > > > > tokens (if SASL is enabled), regular requests
> > > > >
> > > >
> > > >
> > > > Yup!
> > > > "SASL tokens" is a series of proper Kafka protocol
> > SaslHandshakeRequests
> > > > until
> > > > the handshake is done.
> > > >
> > > >
> > > >
> > > > >
> > > > > Is that right? Since we can use either SSL or SASL for
> > authentication,
> > > > it's
> > > > > weird that in one case, we require ApiVersionRequest to happen
> before
> > > > > authentication and in another case we require the reverse.
> > > > >
> > > >
> > > > Since the SSL/TLS is standardised and taken care of for us by the SSL
> > > > libraries it
> > > > doesnt make sense to reimplement that on top of Kafka, so it isn't
> > really
> > > > comparable.
> > > > But for SASL there is no standardised handshake protocol so we must
> > > either
> > > > 

[jira] [Updated] (KAFKA-3600) Enhance java clients to use ApiVersion Req/Resp to check if the broker they are talking to supports required api versions

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3600:
--
Fix Version/s: (was: 0.10.1.0)
   0.10.0.0

> Enhance java clients to use ApiVersion Req/Resp to check if the broker they 
> are talking to supports required api versions
> -
>
> Key: KAFKA-3600
> URL: https://issues.apache.org/jira/browse/KAFKA-3600
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.10.0.0
>
>
> Use the new ApiVersion Req/Resp to enhance java clients to use ApiVersion 
> Req/Resp to check if the broker they are talking to supports required api 
> versions.



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

[jira] [Updated] (KAFKA-3307) Add ApiVersion request/response and server side handling.

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3307:
--
Fix Version/s: 0.10.0.0

> Add ApiVersion request/response and server side handling.
> -
>
> Key: KAFKA-3307
> URL: https://issues.apache.org/jira/browse/KAFKA-3307
> Project: Kafka
>  Issue Type: Sub-task
>Affects Versions: 0.10.0.0
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.10.0.0
>
>




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

[GitHub] kafka pull request: KAFKA-3589: set inner serializer for ChangedSe...

[guozhangwang]

Github user guozhangwang closed the pull request at:

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


---
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-3589) KTable.count(final KeyValueMapper<K, V, K1> selector, String name) throw NPE

[ASF GitHub Bot (JIRA)]

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

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

Github user guozhangwang closed the pull request at:

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


> KTable.count(final KeyValueMapper selector, String name) throw NPE
> 
>
> Key: KAFKA-3589
> URL: https://issues.apache.org/jira/browse/KAFKA-3589
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Damian Guy
>Assignee: Guozhang Wang
>  Labels: architecture
> Fix For: 0.10.0.0
>
>
> The implementation of KTable.count(inal KeyValueMapper selector, 
> String name) passes null through as the parameters for the key and value 
> serdes. This results in an NPE in the aggregate method when it does:
> ChangedSerializer changedValueSerializer = new 
> ChangedSerializer<>(valueSerde.serializer());



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

[jira] [Created] (KAFKA-3604) Improve error messages when null is used with a non-nullable Type

[Grant Henke (JIRA)]

Grant Henke created KAFKA-3604:
--

 Summary: Improve error messages when null is used with a 
non-nullable Type
 Key: KAFKA-3604
 URL: https://issues.apache.org/jira/browse/KAFKA-3604
 Project: Kafka
  Issue Type: Improvement
Affects Versions: 0.10.0.0
Reporter: Grant Henke
Assignee: Grant Henke


Currently when a null is passed to a non-nullable type an unclear message is 
provided in the exception. We should indicate that the issue was caused by a 
null value. 



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

[jira] [Updated] (KAFKA-3600) Enhance java clients to use ApiVersion Req/Resp to check if the broker they are talking to supports required api versions

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3600:
--
Status: Patch Available  (was: Open)

> Enhance java clients to use ApiVersion Req/Resp to check if the broker they 
> are talking to supports required api versions
> -
>
> Key: KAFKA-3600
> URL: https://issues.apache.org/jira/browse/KAFKA-3600
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.10.1.0
>
>
> Use the new ApiVersion Req/Resp to enhance java clients to use ApiVersion 
> Req/Resp to check if the broker they are talking to supports required api 
> versions.



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

[jira] [Resolved] (KAFKA-3589) KTable.count(final KeyValueMapper<K, V, K1> selector, String name) throw NPE

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-3589.
--
Resolution: Fixed

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

> KTable.count(final KeyValueMapper selector, String name) throw NPE
> 
>
> Key: KAFKA-3589
> URL: https://issues.apache.org/jira/browse/KAFKA-3589
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Damian Guy
>Assignee: Guozhang Wang
>  Labels: architecture
> Fix For: 0.10.0.0
>
>
> The implementation of KTable.count(inal KeyValueMapper selector, 
> String name) passes null through as the parameters for the key and value 
> serdes. This results in an NPE in the aggregate method when it does:
> ChangedSerializer changedValueSerializer = new 
> ChangedSerializer<>(valueSerde.serializer());



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

[jira] [Commented] (KAFKA-3601) fail fast when newer client connecting to older server

[Chris Pennello (JIRA)]

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

Chris Pennello commented on KAFKA-3601:
---

Noted.  The brave new world has to start at some version.

> fail fast when newer client connecting to older server
> --
>
> Key: KAFKA-3601
> URL: https://issues.apache.org/jira/browse/KAFKA-3601
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Chris Pennello
>Assignee: Ashish K Singh
>
> I know that connecting with a newer client to an older server is forbidden, 
> but I would like to suggest that the behavior be that it predictably fail 
> noisily, explicitly, and with specific detail indicating why the failure 
> occurred.
> As-is, trying to connect to a v0.8.1.1 cluster with a v0.9.1 client yields a 
> hang when trying to get a coordinator metadata update.
> (This may be related to KAFKA-1894.  I certainly did note 
> {{poll(Long.MAX_VALUE)}} and wept many tears.  At least we could include a 
> TODO-commented constant in the code with a non-forever, timeout, right?)
> {noformat}
>java.lang.Thread.State: RUNNABLE
>   at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
>   at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
>   at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
>   at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
>   - locked <1c8cadab> (a sun.nio.ch.Util$2)
>   - locked <2324ec49> (a java.util.Collections$UnmodifiableSet)
>   - locked <3f3f5e0b> (a sun.nio.ch.EPollSelectorImpl)
>   at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
>   at org.apache.kafka.common.network.Selector.select(Selector.java:425)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:254)
>   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorKnown(AbstractCoordinator.java:184)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:886)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853)
>   ... my code that calls poll...
> {noformat}



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

[jira] [Created] (KAFKA-3603) Define HashCode and Equals methods for Schema, Field and Type

[Grant Henke (JIRA)]

Grant Henke created KAFKA-3603:
--

 Summary: Define HashCode and Equals methods for Schema, Field and 
Type
 Key: KAFKA-3603
 URL: https://issues.apache.org/jira/browse/KAFKA-3603
 Project: Kafka
  Issue Type: Improvement
Reporter: Grant Henke
Assignee: Grant Henke


We should consider implementing HashCode and Equals methods for Schema, Field 
and Type.



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

[jira] [Commented] (KAFKA-3601) fail fast when newer client connecting to older server

[Ashish K Singh (JIRA)]

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

Ashish K Singh commented on KAFKA-3601:
---

[~cpennello_opentable] note that this won't help in case of 0.8.x brokers, as 
the solution requires getting supported api versions from broker. ApiVersion 
request will potentially be added to 0.10.

> fail fast when newer client connecting to older server
> --
>
> Key: KAFKA-3601
> URL: https://issues.apache.org/jira/browse/KAFKA-3601
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Chris Pennello
>Assignee: Ashish K Singh
>
> I know that connecting with a newer client to an older server is forbidden, 
> but I would like to suggest that the behavior be that it predictably fail 
> noisily, explicitly, and with specific detail indicating why the failure 
> occurred.
> As-is, trying to connect to a v0.8.1.1 cluster with a v0.9.1 client yields a 
> hang when trying to get a coordinator metadata update.
> (This may be related to KAFKA-1894.  I certainly did note 
> {{poll(Long.MAX_VALUE)}} and wept many tears.  At least we could include a 
> TODO-commented constant in the code with a non-forever, timeout, right?)
> {noformat}
>java.lang.Thread.State: RUNNABLE
>   at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
>   at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
>   at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
>   at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
>   - locked <1c8cadab> (a sun.nio.ch.Util$2)
>   - locked <2324ec49> (a java.util.Collections$UnmodifiableSet)
>   - locked <3f3f5e0b> (a sun.nio.ch.EPollSelectorImpl)
>   at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
>   at org.apache.kafka.common.network.Selector.select(Selector.java:425)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:254)
>   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorKnown(AbstractCoordinator.java:184)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:886)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853)
>   ... my code that calls poll...
> {noformat}



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

[jira] [Resolved] (KAFKA-3358) Only request metadata updates once we have topics or a pattern subscription

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-3358.

Resolution: Won't Fix

This will be fixed with the TopicMetadata changes in KIP-4.

> Only request metadata updates once we have topics or a pattern subscription
> ---
>
> Key: KAFKA-3358
> URL: https://issues.apache.org/jira/browse/KAFKA-3358
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients
>Affects Versions: 0.9.0.0, 0.9.0.1
>Reporter: Ismael Juma
>Assignee: Jason Gustafson
>Priority: Critical
> Fix For: 0.10.1.0
>
>
> The current code requests a metadata update for _all_ topics which can cause 
> major load issues in large clusters.



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

[jira] [Commented] (KAFKA-3358) Only request metadata updates once we have topics or a pattern subscription

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-3358:


[~ijuma] [~granthenke] I've opened KAFKA-3602 to address the problem with the 
usage of dequeFor(). I'm going to close this issue since it will be fixed with 
the topic metadata changes in KIP-4.

> Only request metadata updates once we have topics or a pattern subscription
> ---
>
> Key: KAFKA-3358
> URL: https://issues.apache.org/jira/browse/KAFKA-3358
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients
>Affects Versions: 0.9.0.0, 0.9.0.1
>Reporter: Ismael Juma
>Assignee: Jason Gustafson
>Priority: Critical
> Fix For: 0.10.1.0
>
>
> The current code requests a metadata update for _all_ topics which can cause 
> major load issues in large clusters.



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

[jira] [Created] (KAFKA-3602) Rename RecordAccumulator dequeueFor() and ensure proper usage

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-3602:
--

 Summary: Rename RecordAccumulator dequeueFor() and ensure proper 
usage
 Key: KAFKA-3602
 URL: https://issues.apache.org/jira/browse/KAFKA-3602
 Project: Kafka
  Issue Type: Bug
Reporter: Jason Gustafson
Assignee: Jason Gustafson


In the investigation of KAFKA-3358, we found a case where the side effect of 
creating record batches in the internal RecordAccumulator.dequeueFor() 
method caused unintended behavior. The bug in this case an implicit assumption 
elsewhere in the code that all record batches correspond to partitions that 
have actually been targeted for sends by the user. This assumption was 
invalidated when a topic metadata request is sent with no topics, which results 
in the metadata from all topics being returned. The end result when that 
happened is that the client gets stuck trying to fetch metadata for topics 
which were not even used.

Although this particular problem will be fixed by changing the TopicMetadata 
request in KIP-4, it probably could have been avoided by making the side-effect 
of batch creation clear in the method name. For example, instead of dequeFor(), 
we should use something like getOrCreateDeque(). It's more verbose, but it 
makes the behavior clear. From a scan of the code, it looks like there are a 
couple places where we do not expect the side-effect of batch creation, so we 
should fix that too.



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

[jira] [Commented] (KAFKA-3601) fail fast when newer client connecting to older server

[Chris Pennello (JIRA)]

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

Chris Pennello commented on KAFKA-3601:
---

Cool, thanks for the context.  (I did search around before opening this issue, 
but didn't find the one you referenced.)

> fail fast when newer client connecting to older server
> --
>
> Key: KAFKA-3601
> URL: https://issues.apache.org/jira/browse/KAFKA-3601
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Chris Pennello
>Assignee: Ashish K Singh
>
> I know that connecting with a newer client to an older server is forbidden, 
> but I would like to suggest that the behavior be that it predictably fail 
> noisily, explicitly, and with specific detail indicating why the failure 
> occurred.
> As-is, trying to connect to a v0.8.1.1 cluster with a v0.9.1 client yields a 
> hang when trying to get a coordinator metadata update.
> (This may be related to KAFKA-1894.  I certainly did note 
> {{poll(Long.MAX_VALUE)}} and wept many tears.  At least we could include a 
> TODO-commented constant in the code with a non-forever, timeout, right?)
> {noformat}
>java.lang.Thread.State: RUNNABLE
>   at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
>   at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
>   at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
>   at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
>   - locked <1c8cadab> (a sun.nio.ch.Util$2)
>   - locked <2324ec49> (a java.util.Collections$UnmodifiableSet)
>   - locked <3f3f5e0b> (a sun.nio.ch.EPollSelectorImpl)
>   at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
>   at org.apache.kafka.common.network.Selector.select(Selector.java:425)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:254)
>   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorKnown(AbstractCoordinator.java:184)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:886)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853)
>   ... my code that calls poll...
> {noformat}



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

[jira] [Commented] (KAFKA-3601) fail fast when newer client connecting to older server

[Ashish K Singh (JIRA)]

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

Ashish K Singh commented on KAFKA-3601:
---

This might be addressed as part of KAFKA-3600.

> fail fast when newer client connecting to older server
> --
>
> Key: KAFKA-3601
> URL: https://issues.apache.org/jira/browse/KAFKA-3601
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Chris Pennello
>Assignee: Ashish K Singh
>
> I know that connecting with a newer client to an older server is forbidden, 
> but I would like to suggest that the behavior be that it predictably fail 
> noisily, explicitly, and with specific detail indicating why the failure 
> occurred.
> As-is, trying to connect to a v0.8.1.1 cluster with a v0.9.1 client yields a 
> hang when trying to get a coordinator metadata update.
> (This may be related to KAFKA-1894.  I certainly did note 
> {{poll(Long.MAX_VALUE)}} and wept many tears.  At least we could include a 
> TODO-commented constant in the code with a non-forever, timeout, right?)
> {noformat}
>java.lang.Thread.State: RUNNABLE
>   at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
>   at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
>   at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
>   at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
>   - locked <1c8cadab> (a sun.nio.ch.Util$2)
>   - locked <2324ec49> (a java.util.Collections$UnmodifiableSet)
>   - locked <3f3f5e0b> (a sun.nio.ch.EPollSelectorImpl)
>   at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
>   at org.apache.kafka.common.network.Selector.select(Selector.java:425)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:254)
>   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorKnown(AbstractCoordinator.java:184)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:886)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853)
>   ... my code that calls poll...
> {noformat}



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

[jira] [Assigned] (KAFKA-3601) fail fast when newer client connecting to older server

[Ashish K Singh (JIRA)]

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

Ashish K Singh reassigned KAFKA-3601:
-

Assignee: Ashish K Singh  (was: Neha Narkhede)

> fail fast when newer client connecting to older server
> --
>
> Key: KAFKA-3601
> URL: https://issues.apache.org/jira/browse/KAFKA-3601
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Chris Pennello
>Assignee: Ashish K Singh
>
> I know that connecting with a newer client to an older server is forbidden, 
> but I would like to suggest that the behavior be that it predictably fail 
> noisily, explicitly, and with specific detail indicating why the failure 
> occurred.
> As-is, trying to connect to a v0.8.1.1 cluster with a v0.9.1 client yields a 
> hang when trying to get a coordinator metadata update.
> (This may be related to KAFKA-1894.  I certainly did note 
> {{poll(Long.MAX_VALUE)}} and wept many tears.  At least we could include a 
> TODO-commented constant in the code with a non-forever, timeout, right?)
> {noformat}
>java.lang.Thread.State: RUNNABLE
>   at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
>   at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
>   at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
>   at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
>   - locked <1c8cadab> (a sun.nio.ch.Util$2)
>   - locked <2324ec49> (a java.util.Collections$UnmodifiableSet)
>   - locked <3f3f5e0b> (a sun.nio.ch.EPollSelectorImpl)
>   at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
>   at org.apache.kafka.common.network.Selector.select(Selector.java:425)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:254)
>   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorKnown(AbstractCoordinator.java:184)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:886)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853)
>   ... my code that calls poll...
> {noformat}



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

[GitHub] kafka pull request: KAFKA-3600: Enhance java clients to use ApiVer...

[SinghAsDev]

GitHub user SinghAsDev opened a pull request:

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

KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to check if the 
broker they are talking to supports required api versions



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

$ git pull https://github.com/SinghAsDev/kafka KAFKA-3600

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

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






---
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-3600) Enhance java clients to use ApiVersion Req/Resp to check if the broker they are talking to supports required api versions

[ASF GitHub Bot (JIRA)]

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

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

GitHub user SinghAsDev opened a pull request:

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

KAFKA-3600: Enhance java clients to use ApiVersion Req/Resp to check if the 
broker they are talking to supports required api versions



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

$ git pull https://github.com/SinghAsDev/kafka KAFKA-3600

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

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






> Enhance java clients to use ApiVersion Req/Resp to check if the broker they 
> are talking to supports required api versions
> -
>
> Key: KAFKA-3600
> URL: https://issues.apache.org/jira/browse/KAFKA-3600
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.10.1.0
>
>
> Use the new ApiVersion Req/Resp to enhance java clients to use ApiVersion 
> Req/Resp to check if the broker they are talking to supports required api 
> versions.



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

[jira] [Created] (KAFKA-3601) fail fast when newer client connecting to older server

[Chris Pennello (JIRA)]

Chris Pennello created KAFKA-3601:
-

 Summary: fail fast when newer client connecting to older server
 Key: KAFKA-3601
 URL: https://issues.apache.org/jira/browse/KAFKA-3601
 Project: Kafka
  Issue Type: Improvement
  Components: consumer
Reporter: Chris Pennello
Assignee: Neha Narkhede


I know that connecting with a newer client to an older server is forbidden, but 
I would like to suggest that the behavior be that it predictably fail noisily, 
explicitly, and with specific detail indicating why the failure occurred.

As-is, trying to connect to a v0.8.1.1 cluster with a v0.9.1 client yields a 
hang when trying to get a coordinator metadata update.

(This may be related to KAFKA-1894.  I certainly did note 
{{poll(Long.MAX_VALUE)}} and wept many tears.  At least we could include a 
TODO-commented constant in the code with a non-forever, timeout, right?)

{noformat}
   java.lang.Thread.State: RUNNABLE
at sun.nio.ch.EPollArrayWrapper.epollWait(Native Method)
at sun.nio.ch.EPollArrayWrapper.poll(EPollArrayWrapper.java:269)
at sun.nio.ch.EPollSelectorImpl.doSelect(EPollSelectorImpl.java:93)
at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:86)
- locked <1c8cadab> (a sun.nio.ch.Util$2)
- locked <2324ec49> (a java.util.Collections$UnmodifiableSet)
- locked <3f3f5e0b> (a sun.nio.ch.EPollSelectorImpl)
at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97)
at org.apache.kafka.common.network.Selector.select(Selector.java:425)
at org.apache.kafka.common.network.Selector.poll(Selector.java:254)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
at 
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
at 
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
at 
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
at 
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:134)
at 
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorKnown(AbstractCoordinator.java:184)
at 
org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:886)
at 
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853)
... my code that calls poll...
{noformat}



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

[jira] [Created] (KAFKA-3600) Enhance java clients to use ApiVersion Req/Resp to check if the broker they are talking to supports required api versions

[Ashish K Singh (JIRA)]

Ashish K Singh created KAFKA-3600:
-

 Summary: Enhance java clients to use ApiVersion Req/Resp to check 
if the broker they are talking to supports required api versions
 Key: KAFKA-3600
 URL: https://issues.apache.org/jira/browse/KAFKA-3600
 Project: Kafka
  Issue Type: Sub-task
Reporter: Ashish K Singh
Assignee: Ashish K Singh


Use the new ApiVersion Req/Resp to enhance java clients to use ApiVersion 
Req/Resp to check if the broker they are talking to supports required api 
versions.



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

[GitHub] kafka pull request: KAFKA-3337: Extract selector as a separate gro...

[asfgit]

Github user asfgit closed the pull request at:

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


---
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-3337) Extract selector as a separate groupBy operator for KTable aggregations

[ASF GitHub Bot (JIRA)]

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

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

Github user asfgit closed the pull request at:

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


> Extract selector as a separate groupBy operator for KTable aggregations
> ---
>
> Key: KAFKA-3337
> URL: https://issues.apache.org/jira/browse/KAFKA-3337
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Matthias J. Sax
>  Labels: api, newbie++
> Fix For: 0.10.0.0
>
>
> Currently KTable aggregation takes a selector used for selecting the 
> aggregate key.and an aggregator for aggregating the values with the same 
> selected key, which makes the function a little bit "heavy":
> {code}
> table.groupBy(initializer, adder, substractor, selector, /* optional serde*/);
> {code}
>  It is better to extract the selector in a separate groupBy function such that
> {code}
> KTableGrouped KTable#groupBy(selector);
> KTable KTableGrouped#aggregate(initializer, adder, substractor, /* optional 
> serde*/);
> {code}
> Note that "KTableGrouped" only have APIs for aggregate and reduce, and none 
> else. So users have to follow the pattern below:
> {code}
> table.groupBy(...).aggregate(...);
> {code}
> This pattern is more natural for users who are familiar with SQL / Pig or 
> Spark DSL, etc.



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

[jira] [Resolved] (KAFKA-3337) Extract selector as a separate groupBy operator for KTable aggregations

[Guozhang Wang (JIRA)]

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

Guozhang Wang resolved KAFKA-3337.
--
Resolution: Fixed

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

> Extract selector as a separate groupBy operator for KTable aggregations
> ---
>
> Key: KAFKA-3337
> URL: https://issues.apache.org/jira/browse/KAFKA-3337
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Matthias J. Sax
>  Labels: api, newbie++
> Fix For: 0.10.0.0
>
>
> Currently KTable aggregation takes a selector used for selecting the 
> aggregate key.and an aggregator for aggregating the values with the same 
> selected key, which makes the function a little bit "heavy":
> {code}
> table.groupBy(initializer, adder, substractor, selector, /* optional serde*/);
> {code}
>  It is better to extract the selector in a separate groupBy function such that
> {code}
> KTableGrouped KTable#groupBy(selector);
> KTable KTableGrouped#aggregate(initializer, adder, substractor, /* optional 
> serde*/);
> {code}
> Note that "KTableGrouped" only have APIs for aggregate and reduce, and none 
> else. So users have to follow the pattern below:
> {code}
> table.groupBy(...).aggregate(...);
> {code}
> This pattern is more natural for users who are familiar with SQL / Pig or 
> Spark DSL, etc.



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

[jira] [Updated] (KAFKA-3307) Add ApiVersion request/response and server side handling.

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3307:
--
Summary: Add ApiVersion request/response and server side handling.  (was: 
Add ProtocolVersion request/response and server side handling.)

> Add ApiVersion request/response and server side handling.
> -
>
> Key: KAFKA-3307
> URL: https://issues.apache.org/jira/browse/KAFKA-3307
> Project: Kafka
>  Issue Type: Sub-task
>Affects Versions: 0.10.0.0
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
>




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

Re: Unknown licenses reported by './gradlew test'

[Guozhang Wang]

Vahid,

I rerun the same command on current trunk but do not reproduce this issue.

I might be that you have some mis testing / util code in the folder, and
although not added to the project, checkstyle still checks ALL files.

Guozhang


On Thu, Apr 21, 2016 at 11:05 AM, Vahid S Hashemian <
vahidhashem...@us.ibm.com> wrote:

> I have started getting this error when running './gradlew test':
>
> kafka@kafka:~/workspace/kafka$ ./gradlew test
> To honour the JVM settings for this build a new JVM will be forked. Please
> consider using the daemon:
> https://docs.gradle.org/2.11/userguide/gradle_daemon.html.
> Building project 'core' with Scala version 2.10.6
> :compileJava UP-TO-DATE
> :processResources UP-TO-DATE
> :classes UP-TO-DATE
> :rat
> Unknown license: /home/kafka/workspace/kafka/bin/gradlew
> Unknown license: /home/kafka/workspace/kafka/bin/gradlew.bat
> :rat FAILED
>
> FAILURE: Build failed with an exception.
>
> * Where:
> Script '/home/kafka/workspace/kafka/gradle/rat.gradle' line: 63
>
> * What went wrong:
> Execution failed for task ':rat'.
> > Found 2 files with unknown licenses.
>
> * Try:
> Run with --stacktrace option to get the stack trace. Run with --info or
> --debug option to get more log output.
>
> BUILD FAILED
>
> Total time: 8.821 secs
>
> Does anyone know what the issue is? Thanks.
>
> Regards,
> Vahid Hashemian
>
>


-- 
-- Guozhang

Unknown licenses reported by './gradlew test'

[Vahid S Hashemian]

I have started getting this error when running './gradlew test':

kafka@kafka:~/workspace/kafka$ ./gradlew test
To honour the JVM settings for this build a new JVM will be forked. Please 
consider using the daemon: 
https://docs.gradle.org/2.11/userguide/gradle_daemon.html.
Building project 'core' with Scala version 2.10.6
:compileJava UP-TO-DATE
:processResources UP-TO-DATE
:classes UP-TO-DATE
:rat
Unknown license: /home/kafka/workspace/kafka/bin/gradlew
Unknown license: /home/kafka/workspace/kafka/bin/gradlew.bat
:rat FAILED

FAILURE: Build failed with an exception.

* Where:
Script '/home/kafka/workspace/kafka/gradle/rat.gradle' line: 63

* What went wrong:
Execution failed for task ':rat'.
> Found 2 files with unknown licenses.

* Try:
Run with --stacktrace option to get the stack trace. Run with --info or 
--debug option to get more log output.

BUILD FAILED

Total time: 8.821 secs

Does anyone know what the issue is? Thanks.

Regards,
Vahid Hashemian

[jira] [Commented] (KAFKA-3565) Producer's throughput lower with compressed data after KIP-31/32

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-3565:
-

[~jkreps] Yes, I agree that "free" calls can make a difference when called 
millions of times. We can try to use a fixed timestamp to verify this. If that 
is the case theoretically we should see a bigger difference if more records are 
sent. That means when snappy is used 0.9 should have a bigger advantage because 
it sends significantly more records per sec, but the cases where 0.9 wins in my 
tests are all using gzip.

Your batching hypothesis may still stand. In all the tests, 0.9 is sending less 
but bigger batches. And trunk is send more but smaller batches. In the cases 
that 0.9 wins, it looks the bigger batch helped. Let me run the test again and 
also get the records per request and compression ratio.

> Producer's throughput lower with compressed data after KIP-31/32
> 
>
> Key: KAFKA-3565
> URL: https://issues.apache.org/jira/browse/KAFKA-3565
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ismael Juma
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> Relative offsets were introduced by KIP-31 so that the broker does not have 
> to recompress data (this was previously required after offsets were 
> assigned). The implicit assumption is that reducing CPU usage required by 
> recompression would mean that producer throughput for compressed data would 
> increase.
> However, this doesn't seem to be the case:
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--012.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   59.030 seconds
> {"records_per_sec": 519418.343653, "mb_per_sec": 49.54}
> {code}
> Full results: https://gist.github.com/ijuma/0afada4ff51ad6a5ac2125714d748292
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--013.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   1 minute 0.243 seconds
> {"records_per_sec": 427308.818848, "mb_per_sec": 40.75}
> {code}
> Full results: https://gist.github.com/ijuma/e49430f0548c4de5691ad47696f5c87d
> The difference for the uncompressed case is smaller (and within what one 
> would expect given the additional size overhead caused by the timestamp 
> field):
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--010.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 4.176 seconds
> {"records_per_sec": 321018.17747, "mb_per_sec": 30.61}
> {code}
> Full results: https://gist.github.com/ijuma/5fec369d686751a2d84debae8f324d4f
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--014.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 5.079 seconds
> {"records_per_sec": 291777.608696, "mb_per_sec": 27.83}
> {code}
> Full results: https://gist.github.com/ijuma/1d35bd831ff9931448b0294bd9b787ed



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

[VOTE] KIP-56 Allow cross origin HTTP requests on all HTTP methods

[Liquan Pei]

Hi

I would like to start vote on KIP-56.

https://cwiki.apache.org/confluence/display/KAFKA/KIP-56%3A+Allow+cross+origin+HTTP+requests+on+all+HTTP+methods

Thanks,

-- 
Liquan Pei
Software Engineer, Confluent Inc

[jira] [Commented] (KAFKA-3579) TopicCommand references outdated consumer property fetch.message.max.bytes

[ASF GitHub Bot (JIRA)]

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

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

GitHub user vahidhashemian reopened a pull request:

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

KAFKA-3579 - Update reference to the outdated consumer property

Replace references to the old consumer property 'fetch.message.max.bytes' 
with the corresponding property in the new consumer 'max.partition.fetch.bytes'.

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

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

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

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


commit fbd579331e8136209356bc572e978908fbde2c1e
Author: Vahid Hashemian 
Date:   2016-04-19T18:21:45Z

KAFKA-3579 - Update reference to the outdated consumer property

Replace references to the old consumer property 'fetch.message.max.bytes' 
with the corresponding property in the new consumer 'max.partition.fetch.bytes'.




> TopicCommand references outdated consumer property fetch.message.max.bytes 
> ---
>
> Key: KAFKA-3579
> URL: https://issues.apache.org/jira/browse/KAFKA-3579
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Jun Rao
>Assignee: Vahid Hashemian
>  Labels: newbie
>
> TopicCommand gives the following warning.
> *
> *** WARNING: you are creating a topic where the the max.message.bytes is 
> greater than the consumer ***
> *** default. This operation is potentially dangerous. Consumers will get 
> failures if their***
> *** fetch.message.max.bytes < the value you are using.
> ***
> *
> - value set here: 130
> - Default Consumer fetch.message.max.bytes: 1048576
> - Default Broker max.message.bytes: 112
> fetch.message.max.bytes is used in the old consumer. We should reference 
> max.partition.fetch.bytes in the new consumer instead.



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

[GitHub] kafka pull request: KAFKA-3579 - Update reference to the outdated ...

[vahidhashemian]

Github user vahidhashemian closed the pull request at:

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


---
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-3579) TopicCommand references outdated consumer property fetch.message.max.bytes

[ASF GitHub Bot (JIRA)]

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

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

Github user vahidhashemian closed the pull request at:

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


> TopicCommand references outdated consumer property fetch.message.max.bytes 
> ---
>
> Key: KAFKA-3579
> URL: https://issues.apache.org/jira/browse/KAFKA-3579
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Jun Rao
>Assignee: Vahid Hashemian
>  Labels: newbie
>
> TopicCommand gives the following warning.
> *
> *** WARNING: you are creating a topic where the the max.message.bytes is 
> greater than the consumer ***
> *** default. This operation is potentially dangerous. Consumers will get 
> failures if their***
> *** fetch.message.max.bytes < the value you are using.
> ***
> *
> - value set here: 130
> - Default Consumer fetch.message.max.bytes: 1048576
> - Default Broker max.message.bytes: 112
> fetch.message.max.bytes is used in the old consumer. We should reference 
> max.partition.fetch.bytes in the new consumer instead.



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

[GitHub] kafka pull request: KAFKA-3579 - Update reference to the outdated ...

[vahidhashemian]

GitHub user vahidhashemian reopened a pull request:

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

KAFKA-3579 - Update reference to the outdated consumer property

Replace references to the old consumer property 'fetch.message.max.bytes' 
with the corresponding property in the new consumer 'max.partition.fetch.bytes'.

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

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

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

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


commit fbd579331e8136209356bc572e978908fbde2c1e
Author: Vahid Hashemian 
Date:   2016-04-19T18:21:45Z

KAFKA-3579 - Update reference to the outdated consumer property

Replace references to the old consumer property 'fetch.message.max.bytes' 
with the corresponding property in the new consumer 'max.partition.fetch.bytes'.




---
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-3565) Producer's throughput lower with compressed data after KIP-31/32

[Jay Kreps (JIRA)]

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

Jay Kreps commented on KAFKA-3565:
--

Oh yeah, I think the critical question is what is the bottleneck? In the cases 
where there is a degradation does it come from the server, the producer I/O 
thread, or the sending thread. I think the producer has metrics that cover this.

> Producer's throughput lower with compressed data after KIP-31/32
> 
>
> Key: KAFKA-3565
> URL: https://issues.apache.org/jira/browse/KAFKA-3565
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ismael Juma
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> Relative offsets were introduced by KIP-31 so that the broker does not have 
> to recompress data (this was previously required after offsets were 
> assigned). The implicit assumption is that reducing CPU usage required by 
> recompression would mean that producer throughput for compressed data would 
> increase.
> However, this doesn't seem to be the case:
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--012.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   59.030 seconds
> {"records_per_sec": 519418.343653, "mb_per_sec": 49.54}
> {code}
> Full results: https://gist.github.com/ijuma/0afada4ff51ad6a5ac2125714d748292
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--013.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   1 minute 0.243 seconds
> {"records_per_sec": 427308.818848, "mb_per_sec": 40.75}
> {code}
> Full results: https://gist.github.com/ijuma/e49430f0548c4de5691ad47696f5c87d
> The difference for the uncompressed case is smaller (and within what one 
> would expect given the additional size overhead caused by the timestamp 
> field):
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--010.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 4.176 seconds
> {"records_per_sec": 321018.17747, "mb_per_sec": 30.61}
> {code}
> Full results: https://gist.github.com/ijuma/5fec369d686751a2d84debae8f324d4f
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--014.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 5.079 seconds
> {"records_per_sec": 291777.608696, "mb_per_sec": 27.83}
> {code}
> Full results: https://gist.github.com/ijuma/1d35bd831ff9931448b0294bd9b787ed



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

[jira] [Commented] (KAFKA-3565) Producer's throughput lower with compressed data after KIP-31/32

[Jay Kreps (JIRA)]

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

Jay Kreps commented on KAFKA-3565:
--

[~becket_qin] Yeah I think you are saying something like "previously a 100 byte 
record had 16 bytes of overhead, now it has 24 so naturally it's slower". The 
reason I find that surprising is because in the 100 byte range we have always 
been bound by the per-record overhead not the raw throughput. You can see this 
in that in this range we max out at around 30MB/sec versus with larger records 
we max out closer to 100MB/sec. But for small records the raw number of them 
per second is crazy high so any per-record work, no matter how small, even 
little things like acquiring a lock or checking the clock, can have outsized 
performance impact since we're trying to do it 800k-1m times per second. We saw 
this last release where we added a few clock checks and it lead to a very 
significant perf drop in the producer. This magnification means adding 
something that is only 100ns to the per-record overhead, which normally is in 
the range of "free", is magnified to something like a 10% overhead in the small 
record regime.

> Producer's throughput lower with compressed data after KIP-31/32
> 
>
> Key: KAFKA-3565
> URL: https://issues.apache.org/jira/browse/KAFKA-3565
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ismael Juma
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> Relative offsets were introduced by KIP-31 so that the broker does not have 
> to recompress data (this was previously required after offsets were 
> assigned). The implicit assumption is that reducing CPU usage required by 
> recompression would mean that producer throughput for compressed data would 
> increase.
> However, this doesn't seem to be the case:
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--012.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   59.030 seconds
> {"records_per_sec": 519418.343653, "mb_per_sec": 49.54}
> {code}
> Full results: https://gist.github.com/ijuma/0afada4ff51ad6a5ac2125714d748292
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--013.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   1 minute 0.243 seconds
> {"records_per_sec": 427308.818848, "mb_per_sec": 40.75}
> {code}
> Full results: https://gist.github.com/ijuma/e49430f0548c4de5691ad47696f5c87d
> The difference for the uncompressed case is smaller (and within what one 
> would expect given the additional size overhead caused by the timestamp 
> field):
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--010.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 4.176 seconds
> {"records_per_sec": 321018.17747, "mb_per_sec": 30.61}
> {code}
> Full results: https://gist.github.com/ijuma/5fec369d686751a2d84debae8f324d4f
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--014.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 5.079 seconds
> {"records_per_sec": 291777.608696, "mb_per_sec": 27.83}
> {code}
> Full results: https://gist.github.com/ijuma/1d35bd831ff9931448b0294bd9b787ed



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

[jira] [Comment Edited] (KAFKA-3565) Producer's throughput lower with compressed data after KIP-31/32

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin edited comment on KAFKA-3565 at 4/21/16 4:09 PM:
--

[~jkreps] The google doc should be accessible now. 

I think the change in uncompressed case is expected because of the 8-bytes 
additional cost. The MB/s reported in ProducerPerformance seems based on the 
actual message size which does not include the message header overhead.

I benchmarked the compression of bounded random integers in the producer with 
jmh:
{noformat}
Benchmark (bufferSize)  (recordSize)  (valueBound)  
Mode  Cnt   ScoreError  Units
ClientMicroBenchMark.compression  1024   100   500  
avgt   20  152745.539 ± 198435.342  ns/op
ClientMicroBenchMark.compression  1024   100  5000  
avgt   20  106698.904 ± 111624.120  ns/op
ClientMicroBenchMark.compression  1024   100 5  
avgt   20  104670.802 ± 110694.704  ns/op
ClientMicroBenchMark.compression  1024  1000   500  
avgt   20  169223.272 ± 180063.271  ns/op
ClientMicroBenchMark.compression  1024  1000  5000  
avgt   20  118949.514 ± 122875.686  ns/op
ClientMicroBenchMark.compression  1024  1000 5  
avgt   20  130193.581 ± 140485.913  ns/op
{noformat}

I benchmarked System.currentTimeMillis() some time ago as well, it took single 
digit nano seconds. So It seems ignorable compared with compression time.

>From [~ijuma] test, it seems the gap is around 8-10% in the throughput of user 
>bytes when message size is 100. Is this different from what we saw before 
>which has 17% gap? What was the integer range of the test? And how many 
>records were sent during the test? One thing I noticed is that it takes a few 
>seconds for the throughput to become stable, so if the tests finished very 
>quickly, the results may not be quite accurate. The tests I ran adjusted the 
>records number dynamically to let the testing time be at least 15 seconds.


was (Author: becket_qin):
[~jkreps] The google doc should be accessible now. 

I think the change in uncompressed case is expected because of the 8-bytes 
additional cost. The MB/s reported in ProducerPerformance seems based on the 
actual message size which does not include the message header overhead.

I benchmarked the compression of bounded random integers in the producer with 
jmh:
{code}
Benchmark (bufferSize)  (recordSize)  (valueBound)  
Mode  Cnt   ScoreError  Units
ClientMicroBenchMark.compression  1024   100   500  
avgt   20  152745.539 ± 198435.342  ns/op
ClientMicroBenchMark.compression  1024   100  5000  
avgt   20  106698.904 ± 111624.120  ns/op
ClientMicroBenchMark.compression  1024   100 5  
avgt   20  104670.802 ± 110694.704  ns/op
ClientMicroBenchMark.compression  1024  1000   500  
avgt   20  169223.272 ± 180063.271  ns/op
ClientMicroBenchMark.compression  1024  1000  5000  
avgt   20  118949.514 ± 122875.686  ns/op
ClientMicroBenchMark.compression  1024  1000 5  
avgt   20  130193.581 ± 140485.913  ns/op
{code}

I benchmarked System.currentTimeMillis() some time ago as well, it took single 
digit nano seconds. So It seems ignorable compared with compression time.

>From [~ijuma] test, it seems the gap is around 8-10% in the throughput of user 
>bytes when message size is 100. Is this different from what we saw before 
>which has 17% gap? What was the integer range of the test? And how many 
>records were sent during the test? One thing I noticed is that it takes a few 
>seconds for the throughput to become stable, so if the tests finished very 
>quickly, the results may not be quite accurate. The tests I ran adjusted the 
>records number dynamically to let the testing time be at least 15 seconds.

> Producer's throughput lower with compressed data after KIP-31/32
> 
>
> Key: KAFKA-3565
> URL: https://issues.apache.org/jira/browse/KAFKA-3565
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ismael Juma
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> Relative offsets were introduced by KIP-31 so that the broker does not have 
> to recompress data (this was previously required after offsets were 
> assigned). The implicit assumption is that reducing CPU usage required by 
> recompression would mean that producer throughput for compressed data would 
> increase.
> However, this doesn't seem to be the case:
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id: 

[jira] [Commented] (KAFKA-3565) Producer's throughput lower with compressed data after KIP-31/32

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-3565:
-

[~jkreps] The google doc should be accessible now. 

I think the change in uncompressed case is expected because of the 8-bytes 
additional cost. The MB/s reported in ProducerPerformance seems based on the 
actual message size which does not include the message header overhead.

I benchmarked the compression of bounded random integers in the producer with 
jmh:
{code}
Benchmark (bufferSize)  (recordSize)  (valueBound)  
Mode  Cnt   ScoreError  Units
ClientMicroBenchMark.compression  1024   100   500  
avgt   20  152745.539 ± 198435.342  ns/op
ClientMicroBenchMark.compression  1024   100  5000  
avgt   20  106698.904 ± 111624.120  ns/op
ClientMicroBenchMark.compression  1024   100 5  
avgt   20  104670.802 ± 110694.704  ns/op
ClientMicroBenchMark.compression  1024  1000   500  
avgt   20  169223.272 ± 180063.271  ns/op
ClientMicroBenchMark.compression  1024  1000  5000  
avgt   20  118949.514 ± 122875.686  ns/op
ClientMicroBenchMark.compression  1024  1000 5  
avgt   20  130193.581 ± 140485.913  ns/op
{code}

I benchmarked System.currentTimeMillis() some time ago as well, it took single 
digit nano seconds. So It seems ignorable compared with compression time.

>From [~ijuma] test, it seems the gap is around 8-10% in the throughput of user 
>bytes when message size is 100. Is this different from what we saw before 
>which has 17% gap? What was the integer range of the test? And how many 
>records were sent during the test? One thing I noticed is that it takes a few 
>seconds for the throughput to become stable, so if the tests finished very 
>quickly, the results may not be quite accurate. The tests I ran adjusted the 
>records number dynamically to let the testing time be at least 15 seconds.

> Producer's throughput lower with compressed data after KIP-31/32
> 
>
> Key: KAFKA-3565
> URL: https://issues.apache.org/jira/browse/KAFKA-3565
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ismael Juma
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> Relative offsets were introduced by KIP-31 so that the broker does not have 
> to recompress data (this was previously required after offsets were 
> assigned). The implicit assumption is that reducing CPU usage required by 
> recompression would mean that producer throughput for compressed data would 
> increase.
> However, this doesn't seem to be the case:
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--012.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   59.030 seconds
> {"records_per_sec": 519418.343653, "mb_per_sec": 49.54}
> {code}
> Full results: https://gist.github.com/ijuma/0afada4ff51ad6a5ac2125714d748292
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--013.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   1 minute 0.243 seconds
> {"records_per_sec": 427308.818848, "mb_per_sec": 40.75}
> {code}
> Full results: https://gist.github.com/ijuma/e49430f0548c4de5691ad47696f5c87d
> The difference for the uncompressed case is smaller (and within what one 
> would expect given the additional size overhead caused by the timestamp 
> field):
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--010.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 4.176 seconds
> {"records_per_sec": 321018.17747, "mb_per_sec": 30.61}
> {code}
> Full results: https://gist.github.com/ijuma/5fec369d686751a2d84debae8f324d4f
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--014.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 5.079 seconds
> {"records_per_sec": 291777.608696, "mb_per_sec": 27.83}
> {code}
> Full results: https://gist.github.com/ijuma/1d35bd831ff9931448b0294bd9b787ed



--
This message 

[jira] [Commented] (KAFKA-3594) Kafka new producer retries doesn't work in 0.9.0.1

[ASF GitHub Bot (JIRA)]

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

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

GitHub user omkreddy reopened a pull request:

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

KAFKA-3594; After calling MemoryRecords.close() method, hasRoomFor() method 
should return false

This exception is occurring when producer is trying to append a record to a 
Re-enqueued record batch in the accumulator. We should not allow to add a 
record to Re-enqueued record batch. This is due a bug in 
MemoryRecords.java/hasRoomFor() method. After calling MemoryRecords.close() 
method, hasRoomFor() method should return false. 

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

$ git pull https://github.com/omkreddy/kafka KAFKA-3594

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

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


commit 294a604114a02414887529d9b9025e11a527344a
Author: Manikumar reddy O 
Date:   2016-04-21T10:24:59Z

KAFKA-3594; After calling MemoryRecords.close() method, hasRooFor() method 
should return false




> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Promise$class.complete(Promise.scala:55) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> 

[GitHub] kafka pull request: KAFKA-3594; After calling MemoryRecords.close(...

[omkreddy]

GitHub user omkreddy reopened a pull request:

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

KAFKA-3594; After calling MemoryRecords.close() method, hasRoomFor() method 
should return false

This exception is occurring when producer is trying to append a record to a 
Re-enqueued record batch in the accumulator. We should not allow to add a 
record to Re-enqueued record batch. This is due a bug in 
MemoryRecords.java/hasRoomFor() method. After calling MemoryRecords.close() 
method, hasRoomFor() method should return false. 

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

$ git pull https://github.com/omkreddy/kafka KAFKA-3594

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

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


commit 294a604114a02414887529d9b9025e11a527344a
Author: Manikumar reddy O 
Date:   2016-04-21T10:24:59Z

KAFKA-3594; After calling MemoryRecords.close() method, hasRooFor() method 
should return false




---
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-3594) Kafka new producer retries doesn't work in 0.9.0.1

[ASF GitHub Bot (JIRA)]

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

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

Github user omkreddy closed the pull request at:

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


> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Promise$class.complete(Promise.scala:55) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:67)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:82)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch.run(BatchingExecutor.scala:58) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
>  

[GitHub] kafka pull request: KAFKA-3594; After calling MemoryRecords.close(...

[omkreddy]

Github user omkreddy closed the pull request at:

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


---
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-3565) Producer's throughput lower with compressed data after KIP-31/32

[Jay Kreps (JIRA)]

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

Jay Kreps commented on KAFKA-3565:
--

[~ijuma] Yeah your data eliminates my guess. If the problem were the change in 
batching you wouldn't see impact in the non-compressed case.

[~becket_qin] I don't think we have access to that google doc, but I'm not sure 
how intuitive that is to me, maybe you can explain it more? For one thing the 
timestamps themselves should be ultra compressible, how much bigger is the 
resulting log with timestamps? Also, can we look at actual hprof output for one 
case where there is a degradation and see which methods the change comes from 
and confirm it is consistent with that theory?

One thing we noted in the last release was that System.currentTimeMillis() is 
pretty expensive if it is in the critical path of send(). I wonder if the 
difference remains if we manually set the timestamp to some fixed value and 
don't do it dynamically with the system call?

> Producer's throughput lower with compressed data after KIP-31/32
> 
>
> Key: KAFKA-3565
> URL: https://issues.apache.org/jira/browse/KAFKA-3565
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ismael Juma
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> Relative offsets were introduced by KIP-31 so that the broker does not have 
> to recompress data (this was previously required after offsets were 
> assigned). The implicit assumption is that reducing CPU usage required by 
> recompression would mean that producer throughput for compressed data would 
> increase.
> However, this doesn't seem to be the case:
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--012.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   59.030 seconds
> {"records_per_sec": 519418.343653, "mb_per_sec": 49.54}
> {code}
> Full results: https://gist.github.com/ijuma/0afada4ff51ad6a5ac2125714d748292
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--013.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   1 minute 0.243 seconds
> {"records_per_sec": 427308.818848, "mb_per_sec": 40.75}
> {code}
> Full results: https://gist.github.com/ijuma/e49430f0548c4de5691ad47696f5c87d
> The difference for the uncompressed case is smaller (and within what one 
> would expect given the additional size overhead caused by the timestamp 
> field):
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--010.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 4.176 seconds
> {"records_per_sec": 321018.17747, "mb_per_sec": 30.61}
> {code}
> Full results: https://gist.github.com/ijuma/5fec369d686751a2d84debae8f324d4f
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--014.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 5.079 seconds
> {"records_per_sec": 291777.608696, "mb_per_sec": 27.83}
> {code}
> Full results: https://gist.github.com/ijuma/1d35bd831ff9931448b0294bd9b787ed



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

[jira] [Updated] (KAFKA-3128) Add metrics for ZooKeeper events

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3128:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.0.0

> Add metrics for ZooKeeper events
> 
>
> Key: KAFKA-3128
> URL: https://issues.apache.org/jira/browse/KAFKA-3128
> Project: Kafka
>  Issue Type: Improvement
>  Components: core, zkclient
>Reporter: Flavio Junqueira
> Fix For: 0.10.0.0
>
>
> It would be useful to report via Kafka metrics the number of ZK event 
> notifications, such as connection loss events, session expiration events, 
> etc., as a way of spotting potential issues with the communication with the 
> ZK ensemble.



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

[jira] [Assigned] (KAFKA-3128) Add metrics for ZooKeeper events

[Ismael Juma (JIRA)]

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

Ismael Juma reassigned KAFKA-3128:
--

Assignee: Ismael Juma

> Add metrics for ZooKeeper events
> 
>
> Key: KAFKA-3128
> URL: https://issues.apache.org/jira/browse/KAFKA-3128
> Project: Kafka
>  Issue Type: Improvement
>  Components: core, zkclient
>Reporter: Flavio Junqueira
>Assignee: Ismael Juma
> Fix For: 0.10.0.0
>
>
> It would be useful to report via Kafka metrics the number of ZK event 
> notifications, such as connection loss events, session expiration events, 
> etc., as a way of spotting potential issues with the communication with the 
> ZK ensemble.



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

Kafka Streams - read topic from beginning

[Maria Abramiuc]

Kafka Streams look great, but there is one thing I don't seem to find a way
to do:

- read a topic from beginning even if there is a offset saved:

 I have :

 props.put(StreamsConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

this works as described if there is no offset save.

  For a normal consumer we have:

  seekToBeginning


In KafkaConsumer:

Line 132:

NetworkClient netClient = new NetworkClient(new
Selector(config.getLong("connections.max.idle.ms").longValue(),
this.metrics, this.time, metricGrpPrefix, metricsTags,
channelBuilder), this.metadata, this.clientId, 100,
config.getLong("reconnect.backoff.ms").longValue(),
config.getInt("send.buffer.bytes").intValue(),
config.getInt("receive.buffer.bytes").intValue(),
config.getInt("request.timeout.ms").intValue(), this.time);
this.client = new ConsumerNetworkClient(netClient, this.metadata,
this.time, this.retryBackoffMs);
OffsetResetStrategy offsetResetStrategy =
OffsetResetStrategy.valueOf(config.getString("auto.offset.reset").toUpperCase());
this.subscriptions = new SubscriptionState(offsetResetStrategy);

I can't find any way to set the consumer using StreamsConfig
properties to seekToBeginning or to set subscriptionState to need
offset reset.



*Is there a way to force the consumption of a topic from begging using
Kafka Streams?*


Thank you for all the support provided,

Maria Abramiuc

[jira] [Commented] (KAFKA-3599) Minor code cleanup

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax commented on KAFKA-3599:


[~guozhang] if I understand correctly, we remove {{equalsTo()}} and just use 
{{equals()}}. I also discovered, that {{Windows}} (plural, not singular) has 
{{equalsTo()}} as well, but the method is never used. For consistency, it might 
be good to remove {{equalsTo()}} from {{Windows}} completely or rename it to 
{{equals()}}? What do you think about it?

> Minor code cleanup
> --
>
> Key: KAFKA-3599
> URL: https://issues.apache.org/jira/browse/KAFKA-3599
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Minor
>  Labels: api
>
> - Clean up {{Window.equals()}} and {{Window.equalsTo()}}. Both do the same.
> - remove {{WindowStoreUtils}} from public API, ie, move to sub-package 
> {{internals}}



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

[jira] [Work started] (KAFKA-3599) Minor code cleanup

[Matthias J. Sax (JIRA)]

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

Work on KAFKA-3599 started by Matthias J. Sax.
--
> Minor code cleanup
> --
>
> Key: KAFKA-3599
> URL: https://issues.apache.org/jira/browse/KAFKA-3599
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Minor
>  Labels: api
>
> - Clean up {{Window.equals()}} and {{Window.equalsTo()}}. Both do the same.
> - remove {{WindowStoreUtils}} from public API, ie, move to sub-package 
> {{internals}}



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

[jira] [Created] (KAFKA-3599) Minor code cleanup

[Matthias J. Sax (JIRA)]

Matthias J. Sax created KAFKA-3599:
--

 Summary: Minor code cleanup
 Key: KAFKA-3599
 URL: https://issues.apache.org/jira/browse/KAFKA-3599
 Project: Kafka
  Issue Type: Sub-task
  Components: streams
Reporter: Matthias J. Sax
Assignee: Matthias J. Sax
Priority: Minor


- Clean up {{Window.equals()}} and {{Window.equalsTo()}}. Both do the same.
- remove {{WindowStoreUtils}} from public API, ie, move to sub-package 
{{internals}}




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

[jira] [Work started] (KAFKA-3598) Improve JavaDoc of public API

[Matthias J. Sax (JIRA)]

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

Work on KAFKA-3598 started by Matthias J. Sax.
--
> Improve JavaDoc of public API
> -
>
> Key: KAFKA-3598
> URL: https://issues.apache.org/jira/browse/KAFKA-3598
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Trivial
>
> Add missing JavaDoc to all {{public}} methods of public API. Is related to 
> KAFKA-3440 and KAFKA-3574.



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

[GitHub] kafka pull request: KAFKA-3598: Improve JavaDoc of public API

[mjsax]

GitHub user mjsax opened a pull request:

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

KAFKA-3598: Improve JavaDoc of public API



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

$ git pull https://github.com/mjsax/kafka JavaDoc-publicAPI

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

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


commit 3dcbebb324777e5f6f5ea1df490f1af20c77038f
Author: Matthias J. Sax 
Date:   2016-04-18T19:15:23Z

KAFKA-3598: Improve JavaDoc of public API




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

[jira] [Commented] (KAFKA-3598) Improve JavaDoc of public API

[ASF GitHub Bot (JIRA)]

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

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

GitHub user mjsax opened a pull request:

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

KAFKA-3598: Improve JavaDoc of public API



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

$ git pull https://github.com/mjsax/kafka JavaDoc-publicAPI

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

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


commit 3dcbebb324777e5f6f5ea1df490f1af20c77038f
Author: Matthias J. Sax 
Date:   2016-04-18T19:15:23Z

KAFKA-3598: Improve JavaDoc of public API




> Improve JavaDoc of public API
> -
>
> Key: KAFKA-3598
> URL: https://issues.apache.org/jira/browse/KAFKA-3598
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Trivial
>
> Add missing JavaDoc to all {{public}} methods of public API. Is related to 
> KAFKA-3440 and KAFKA-3574.



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

[jira] [Updated] (KAFKA-3598) Improve JavaDoc of public API

[Matthias J. Sax (JIRA)]

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

Matthias J. Sax updated KAFKA-3598:
---
Description: Add missing JavaDoc to all {{public}} methods of public API. 
Is related to KAFKA-3440 and KAFKA-3574.  (was: Add missing JavaDoc to all 
``public`` methods of public API.)

> Improve JavaDoc of public API
> -
>
> Key: KAFKA-3598
> URL: https://issues.apache.org/jira/browse/KAFKA-3598
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Matthias J. Sax
>Assignee: Matthias J. Sax
>Priority: Trivial
>
> Add missing JavaDoc to all {{public}} methods of public API. Is related to 
> KAFKA-3440 and KAFKA-3574.



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

[jira] [Created] (KAFKA-3598) Improve JavaDoc of public API

[Matthias J. Sax (JIRA)]

Matthias J. Sax created KAFKA-3598:
--

 Summary: Improve JavaDoc of public API
 Key: KAFKA-3598
 URL: https://issues.apache.org/jira/browse/KAFKA-3598
 Project: Kafka
  Issue Type: Sub-task
  Components: streams
Reporter: Matthias J. Sax
Assignee: Matthias J. Sax
Priority: Trivial


Add missing JavaDoc to all ``public`` methods of public API.



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

[jira] [Updated] (KAFKA-3568) KafkaProducer fails with timeout on message send()

[Ismael Juma (JIRA)]

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

Ismael Juma updated KAFKA-3568:
---
Fix Version/s: (was: 0.10.0.0)
   (was: 0.10.1.0)

> KafkaProducer fails with timeout on message send()
> --
>
> Key: KAFKA-3568
> URL: https://issues.apache.org/jira/browse/KAFKA-3568
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.1.0
> Environment: MacOS Docker
>Reporter: Greg Zoller
>  Labels: producer
>
> I had a KafkaProducer working fine in 0.9.0.1.  I was having unrelated 
> problems in that version so thought to try 0.10.1.0.  I built it as I did 
> 0.9.0.1:
> Fresh build against Scala 2.11.7.  Built the tgz build plus local maven 
> install.  From the tgz I created a Docker image similar to spotify/kafka.  I 
> linked my producer code to the maven jars.  This process worked in 0.9.
> Code is here:  
> https://gist.github.com/gzoller/145faef1fefc8acea212e87e06fc86e8
> At the bottom you can see a clip from the output... there's a warning about 
> metadata (not sure if its important or not) and then its trying to send() 
> messages and timing out.  I clipped the output, but it does fail the same way 
> for each message sent in 0.10.1.0.  Same code compiled against 0.9.0.1 
> populates the topic's partitions w/o problem.
> Was there a breaking change between 0.9 and 0.10, or is this a bug?



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

[jira] [Updated] (KAFKA-3594) Kafka new producer retries doesn't work in 0.9.0.1

[Manikumar Reddy (JIRA)]

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

Manikumar Reddy updated KAFKA-3594:
---
Status: Patch Available  (was: Open)

> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Promise$class.complete(Promise.scala:55) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:67)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:82)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch.run(BatchingExecutor.scala:58) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:41) 
> [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:393)
>  [com.typesafe.akka.akka-actor_2.11-2.3.4.jar:na]
> at 
> 

[GitHub] kafka pull request: KAFKA-3594; After calling MemoryRecords.close(...

[omkreddy]

GitHub user omkreddy opened a pull request:

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

KAFKA-3594; After calling MemoryRecords.close() method, hasRooFor() method 
should return false

This exception is occurring when producer is trying to append a record to a 
Re-enqueued record batch in the accumulator. We should not allow to add a 
record to Re-enqueued record batch. This is due a bug in 
MemoryRecords.java/hasRooFor() method. After calling MemoryRecords.close() 
method, hasRooFor() method should return false. 

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

$ git pull https://github.com/omkreddy/kafka KAFKA-3594

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

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


commit 3f018abf2b7b570e220ffab285863eb4b8a55326
Author: Manikumar reddy O 
Date:   2016-04-21T10:24:59Z

KAFKA-3594; After calling MemoryRecords.close() method, hasRooFor() method 
should return false




---
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-3594) Kafka new producer retries doesn't work in 0.9.0.1

[ASF GitHub Bot (JIRA)]

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

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

GitHub user omkreddy opened a pull request:

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

KAFKA-3594; After calling MemoryRecords.close() method, hasRooFor() method 
should return false

This exception is occurring when producer is trying to append a record to a 
Re-enqueued record batch in the accumulator. We should not allow to add a 
record to Re-enqueued record batch. This is due a bug in 
MemoryRecords.java/hasRooFor() method. After calling MemoryRecords.close() 
method, hasRooFor() method should return false. 

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

$ git pull https://github.com/omkreddy/kafka KAFKA-3594

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

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


commit 3f018abf2b7b570e220ffab285863eb4b8a55326
Author: Manikumar reddy O 
Date:   2016-04-21T10:24:59Z

KAFKA-3594; After calling MemoryRecords.close() method, hasRooFor() method 
should return false




> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Promise$class.complete(Promise.scala:55) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> 

[jira] [Comment Edited] (KAFKA-3594) Kafka new producer retries doesn't work in 0.9.0.1

[Manikumar Reddy (JIRA)]

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

Manikumar Reddy edited comment on KAFKA-3594 at 4/21/16 10:43 AM:
--

Thanks for reporting this. I reproduced the error by creating producer retry 
scenario. 
Below exception is occurring when producer is trying to append a record to a 
Re-enqueued record batch in the accumulator.  We should not allow to add a 
record to Re-enqueued record batch. This is happening due a bug in 
MemoryRecords.java/hasRooFor() method. After calling MemoryRecords.close() 
method, hasRooFor() method should return false. I will submit patch soon.

{code}
java.lang.IllegalStateException: Memory records is not writable
at 
org.apache.kafka.common.record.MemoryRecords.append(MemoryRecords.java:95)
at 
org.apache.kafka.clients.producer.internals.RecordBatch.tryAppend(RecordBatch.java:70)
at 
org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:176)
at 
org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:467)
at 
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:430)
{code}


was (Author: omkreddy):
Thanks for reporting this. I reproduced the error by creating producer retry 
scenario. 
Below exception is occurring when producer is trying to append a record to a 
Re-enqueued record batch in the accumulator.
We should not allow to add a record to Re-enqueued record batch. This is 
happening due a bug in MemoryRecords.java/hasRooFor()
method. After calling MemoryRecords.close() method, hasRooFor() method should 
return false. I will submit patch soon.

{code}
java.lang.IllegalStateException: Memory records is not writable
at 
org.apache.kafka.common.record.MemoryRecords.append(MemoryRecords.java:95)
at 
org.apache.kafka.clients.producer.internals.RecordBatch.tryAppend(RecordBatch.java:70)
at 
org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:176)
at 
org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:467)
at 
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:430)
{code}

> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> 

[jira] [Updated] (KAFKA-3594) Kafka new producer retries doesn't work in 0.9.0.1

[Manikumar Reddy (JIRA)]

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

Manikumar Reddy updated KAFKA-3594:
---
 Assignee: Manikumar Reddy  (was: Jun Rao)
 Priority: Critical  (was: Major)
Fix Version/s: 0.10.0.0

Thanks for reporting this. I reproduced the error by creating producer retry 
scenario. 
Below exception is occurring when producer is trying to append a record to a 
Re-enqueued record batch in the accumulator.
We should not allow to add a record to Re-enqueued record batch. This is 
happening due a bug in MemoryRecords.java/hasRooFor()
method. After calling MemoryRecords.close() method, hasRooFor() method should 
return false. I will submit patch soon.

{code}
java.lang.IllegalStateException: Memory records is not writable
at 
org.apache.kafka.common.record.MemoryRecords.append(MemoryRecords.java:95)
at 
org.apache.kafka.clients.producer.internals.RecordBatch.tryAppend(RecordBatch.java:70)
at 
org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:176)
at 
org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:467)
at 
org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:430)
{code}

> Kafka new producer retries doesn't work in 0.9.0.1
> --
>
> Key: KAFKA-3594
> URL: https://issues.apache.org/jira/browse/KAFKA-3594
> Project: Kafka
>  Issue Type: Bug
>  Components: producer 
>Affects Versions: 0.9.0.1
> Environment: Debian 7.8 Wheezy / Kafka 0.9.0.1 (Confluent 2.0.1)
>Reporter: Nicolas PHUNG
>Assignee: Manikumar Reddy
>Priority: Critical
>  Labels: kafka, new, producer, replication, retry
> Fix For: 0.10.0.0
>
>
> Hello,
> I'm encountering an issue with the new Producer on 0.9.0.1 client with a 
> 0.9.0.1 Kafka broker when Kafka broker are offline for example. It seems the 
> retries doesn't work anymore and I got the following error logs :
> {noformat}
> play.api.Application$$anon$1: Execution exception[[IllegalStateException: 
> Memory records is not writable]]
> at play.api.Application$class.handleError(Application.scala:296) 
> ~[com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at play.api.DefaultApplication.handleError(Application.scala:402) 
> [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3$$anonfun$applyOrElse$4.apply(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at scala.Option.map(Option.scala:146) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:320)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:316)
>  [com.typesafe.play.play_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:346) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:345) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:46) 
> [com.typesafe.play.play-iteratees_2.11-2.3.10.jar:2.3.10]
> at 
> scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Promise$class.complete(Promise.scala:55) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:237) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) 
> [org.scala-lang.scala-library-2.11.8.jar:na]
> at 
> akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:67)
>  

[jira] [Commented] (KAFKA-3589) KTable.count(final KeyValueMapper<K, V, K1> selector, String name) throw NPE

[Damian Guy (JIRA)]

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

Damian Guy commented on KAFKA-3589:
---

LGTM

> KTable.count(final KeyValueMapper selector, String name) throw NPE
> 
>
> Key: KAFKA-3589
> URL: https://issues.apache.org/jira/browse/KAFKA-3589
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Damian Guy
>Assignee: Guozhang Wang
>  Labels: architecture
> Fix For: 0.10.0.0
>
>
> The implementation of KTable.count(inal KeyValueMapper selector, 
> String name) passes null through as the parameters for the key and value 
> serdes. This results in an NPE in the aggregate method when it does:
> ChangedSerializer changedValueSerializer = new 
> ChangedSerializer<>(valueSerde.serializer());



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

[jira] [Commented] (KAFKA-3565) Producer's throughput lower with compressed data after KIP-31/32

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-3565:
-

[~ijuma] [~junrao] [~jkreps]

I just finished the parameterized test on the code with and without KIP31/32. 
Here is the result:
https://docs.google.com/spreadsheets/d/1DR13ng6ZMRFki9QCepvsDcT2hfE1sjodql9zR_y2XKs/edit?usp=sharing

A few explanation on the test.

The parameters I used are the following:
max.in.flight.requests.per.connection=1, 5
valueBound=500, 5000, 5
linger.ms=0, 10, 100
recordSize=100, 1000
compression.type=gzip, snappy

*The tests ran with 1 partition, 1 replica, batch.size=50, acks=1. 
*The 0.9 test broker and the trunk test broker were running on the same 
standalone machine. The machine was almost idle except for running the two 
brokers.
*The producers were running on another test box. The RTT between producer and 
broker is negligible (the two machines are only 3 feet away from each other)
*The test runs sequentially, i.e. there is only one producer running during 
each test.

In the test result table, the result of the trunk comes first. Out of all the 
72 runs, 0.9 wins in the following configuration combinations:
1. max.in.flight.requests.per.connection=5, valueBound=500, linger.ms=0, 
messageSize=100, compression.type=gzip
2. max.in.flight.requests.per.connection=5, valueBound=500, linger.ms=0, 
messageSize=1000, compression.type=gzip
3. max.in.flight.requests.per.connection=5, valueBound=500, linger.ms=10, 
messageSize=100, compression.type=gzip
4. max.in.flight.requests.per.connection=5, valueBound=500, linger.ms=10, 
messageSize=1000, compression.type=gzip
5. max.in.flight.requests.per.connection=5, valueBound=500, linger.ms=100, 
messageSize=1000, compression.type=gzip
6. max.in.flight.requests.per.connection=5, valueBound=5000, linger.ms=100, 
messageSize=100, compression.type=gzip

The common thing about these combination is that the valueBound is small. This 
seems indicating that the more compressible the message is the bigger negative 
impact would the 8-bytes overhead have.

There are many other interesting things can be seen from the detailed metrics 
in the table. But for now with the tests, it seems there is no obvious 
unexpected performance issue with KIP-31/32. Please let me know if you think 
there is something I missed. Thanks.


> Producer's throughput lower with compressed data after KIP-31/32
> 
>
> Key: KAFKA-3565
> URL: https://issues.apache.org/jira/browse/KAFKA-3565
> Project: Kafka
>  Issue Type: Bug
>Reporter: Ismael Juma
>Priority: Critical
> Fix For: 0.10.0.0
>
>
> Relative offsets were introduced by KIP-31 so that the broker does not have 
> to recompress data (this was previously required after offsets were 
> assigned). The implicit assumption is that reducing CPU usage required by 
> recompression would mean that producer throughput for compressed data would 
> increase.
> However, this doesn't seem to be the case:
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--012.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   59.030 seconds
> {"records_per_sec": 519418.343653, "mb_per_sec": 49.54}
> {code}
> Full results: https://gist.github.com/ijuma/0afada4ff51ad6a5ac2125714d748292
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
> 2016-04-15--013.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100.compression_type=snappy
> status: PASS
> run time:   1 minute 0.243 seconds
> {"records_per_sec": 427308.818848, "mb_per_sec": 40.75}
> {code}
> Full results: https://gist.github.com/ijuma/e49430f0548c4de5691ad47696f5c87d
> The difference for the uncompressed case is smaller (and within what one 
> would expect given the additional size overhead caused by the timestamp 
> field):
> {code}
> Commit: eee95228fabe1643baa016a2d49fb0a9fe2c66bd (one before KIP-31/32)
> test_id:
> 2016-04-15--010.kafkatest.tests.benchmark_test.Benchmark.test_producer_throughput.topic=topic-replication-factor-three.security_protocol=PLAINTEXT.acks=1.message_size=100
> status: PASS
> run time:   1 minute 4.176 seconds
> {"records_per_sec": 321018.17747, "mb_per_sec": 30.61}
> {code}
> Full results: https://gist.github.com/ijuma/5fec369d686751a2d84debae8f324d4f
> {code}
> Commit: fa594c811e4e329b6e7b897bce910c6772c46c0f (KIP-31/32)
> test_id:
>