[jira] [Commented] (KAFKA-8183) Trogdor - ProduceBench should retry on UnknownTopicOrPartitionException during topic creation
[ https://issues.apache.org/jira/browse/KAFKA-8183?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16863827#comment-16863827 ] Stanislav Kozlovski commented on KAFKA-8183: [https://github.com/apache/kafka/pull/6937] should fix the issue > Trogdor - ProduceBench should retry on UnknownTopicOrPartitionException > during topic creation > - > > Key: KAFKA-8183 > URL: https://issues.apache.org/jira/browse/KAFKA-8183 > Project: Kafka > Issue Type: Improvement >Reporter: Stanislav Kozlovski >Assignee: Stanislav Kozlovski >Priority: Minor > > There exists a race condition in the Trogdor produce bench worker code where > `WorkerUtils#createTopics()` [notices the topic > exists|https://github.com/apache/kafka/blob/4824dc994d7fc56b7540b643a78aadb4bdd0f14d/tools/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java#L159] > yet when it goes on to verify the topics, the DescribeTopics call throws an > `UnknownTopicOrPartitionException`. > We should add sufficient retries such that this does not fail the task. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-8526) Broker may select a failed dir for new replica even in the presence of other live dirs
[ https://issues.apache.org/jira/browse/KAFKA-8526?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Igor Soarez reassigned KAFKA-8526: -- Assignee: Igor Soarez > Broker may select a failed dir for new replica even in the presence of other > live dirs > -- > > Key: KAFKA-8526 > URL: https://issues.apache.org/jira/browse/KAFKA-8526 > Project: Kafka > Issue Type: Bug >Affects Versions: 1.1.1, 2.0.1, 2.1.1, 2.3.0, 2.2.1 >Reporter: Anna Povzner >Assignee: Igor Soarez >Priority: Major > > Suppose a broker is configured with multiple log dirs. One of the log dirs > fails, but there is no load on that dir, so the broker does not know about > the failure yet, _i.e._, the failed dir is still in LogManager#_liveLogDirs. > Suppose a new topic gets created, and the controller chooses the broker with > failed log dir to host one of the replicas. The broker gets LeaderAndIsr > request with isNew flag set. LogManager#getOrCreateLog() selects a log dir > for the new replica from _liveLogDirs, then one two things can happen: > 1) getAbsolutePath can fail, in which case getOrCreateLog will throw an > IOException > 2) Creating directory for new the replica log may fail (_e.g._, if directory > becomes read-only, so getAbsolutePath worked). > In both cases, the selected dir will be marked offline (which is correct). > However, LeaderAndIsr will return an error and replica will be marked > offline, even though the broker may have other live dirs. > *Proposed solution*: Broker should retry selecting a dir for the new replica, > if initially selected dir threw an IOException when trying to create a > directory for the new replica. We should be able to do that in > LogManager#getOrCreateLog() method, but keep in mind that > logDirFailureChannel.maybeAddOfflineLogDir does not synchronously removes the > dir from _liveLogDirs. So, it makes sense to select initial dir by calling > LogManager#nextLogDir (current implementation), but if we fail to create log > on that dir, one approach is to select next dir from _liveLogDirs in > round-robin fashion (until we get to initial log dir – the case where all > dirs failed). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8532) controller-event-thread deadlock with zk-session-expiry-handler0
[ https://issues.apache.org/jira/browse/KAFKA-8532?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864136#comment-16864136 ] Ismael Juma commented on KAFKA-8532: Thanks for the report. Can you reproduce this with 2.2.1 or trunk? > controller-event-thread deadlock with zk-session-expiry-handler0 > > > Key: KAFKA-8532 > URL: https://issues.apache.org/jira/browse/KAFKA-8532 > Project: Kafka > Issue Type: Bug > Components: core >Affects Versions: 2.1.1 >Reporter: leibo >Priority: Blocker > Attachments: js.log > > > We have observed a serious deadlock between controller-event-thead and > zk-session-expirey-handle thread. When this issue occurred, it's only one way > to recovery the kafka cluster is restart kafka server. The follows is the > jstack log of controller-event-thead and zk-session-expiry-handle thread. > "zk-session-expiry-handler0" #163089 daemon prio=5 os_prio=0 > tid=0x7fcc9c01 nid=0xfb22 waiting on condition [0x7fcbb01f8000] > java.lang.Thread.State: WAITING (parking) > at sun.misc.Unsafe.park(Native Method) > - parking to wait for <0x0005ee3f7000> (a > java.util.concurrent.CountDownLatch$Sync) > at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) > at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) // > 等待controller-event-thread线程处理expireEvent > at > kafka.controller.KafkaController$Expire.waitUntilProcessingStarted(KafkaController.scala:1533) > at > kafka.controller.KafkaController$$anon$7.beforeInitializingSession(KafkaController.scala:173) > at > kafka.zookeeper.ZooKeeperClient.callBeforeInitializingSession(ZooKeeperClient.scala:408) > at > kafka.zookeeper.ZooKeeperClient.$anonfun$reinitialize$1(ZooKeeperClient.scala:374) > at > kafka.zookeeper.ZooKeeperClient.$anonfun$reinitialize$1$adapted(ZooKeeperClient.scala:374) > at kafka.zookeeper.ZooKeeperClient$$Lambda$1473/1823438251.apply(Unknown > Source) > at scala.collection.Iterator.foreach(Iterator.scala:937) > at scala.collection.Iterator.foreach$(Iterator.scala:937) > at scala.collection.AbstractIterator.foreach(Iterator.scala:1425) > at scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:209) > at kafka.zookeeper.ZooKeeperClient.reinitialize(ZooKeeperClient.scala:374) > at > kafka.zookeeper.ZooKeeperClient.$anonfun$scheduleSessionExpiryHandler$1(ZooKeeperClient.scala:428) > at > kafka.zookeeper.ZooKeeperClient$$Lambda$1471/701792920.apply$mcV$sp(Unknown > Source) > at kafka.utils.KafkaScheduler.$anonfun$schedule$2(KafkaScheduler.scala:114) > at kafka.utils.KafkaScheduler$$Lambda$198/1048098469.apply$mcV$sp(Unknown > Source) > at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:63) > at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) > at java.util.concurrent.FutureTask.run(FutureTask.java:266) > at > java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) > at > java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) > at > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > at > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > at java.lang.Thread.run(Thread.java:748) > Locked ownable synchronizers: > - <0x000661e8d2e0> (a java.util.concurrent.ThreadPoolExecutor$Worker) > "controller-event-thread" #51 prio=5 os_prio=0 tid=0x7fceaeec4000 > nid=0x310 waiting on condition [0x7fccb55c8000] > java.lang.Thread.State: WAITING (parking) > at sun.misc.Unsafe.park(Native Method) > - parking to wait for <0x0005d1be5a00> (a > java.util.concurrent.CountDownLatch$Sync) > at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) > at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) > at kafka.zookeeper.ZooKeeperClient.handleRequests(ZooKeeperClient.scala:157) > at > kafka.zk.KafkaZkClient.re
[jira] [Resolved] (KAFKA-6874) Add Configuration Allowing for Optional Topology Optimization
[ https://issues.apache.org/jira/browse/KAFKA-6874?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bill Bejeck resolved KAFKA-6874. Resolution: Duplicate Fix Version/s: 2.0.0 Duplicate of https://issues.apache.org/jira/browse/KAFKA-6935 Resolved with PR [https://github.com/apache/kafka/pull/5071] > Add Configuration Allowing for Optional Topology Optimization > -- > > Key: KAFKA-6874 > URL: https://issues.apache.org/jira/browse/KAFKA-6874 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Bill Bejeck >Assignee: Bill Bejeck >Priority: Major > Fix For: 2.0.0 > > > With the release of 2.0 Streams will introduce topology optimization. We > should provide a config with a default value of false allowing users to > enable/disable optimization. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-6874) Add Configuration Allowing for Optional Topology Optimization
[ https://issues.apache.org/jira/browse/KAFKA-6874?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864155#comment-16864155 ] Bill Bejeck commented on KAFKA-6874: [~mjsax] resolved as duplicate, let me know if there is anything else to do. > Add Configuration Allowing for Optional Topology Optimization > -- > > Key: KAFKA-6874 > URL: https://issues.apache.org/jira/browse/KAFKA-6874 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: Bill Bejeck >Assignee: Bill Bejeck >Priority: Major > Fix For: 2.0.0 > > > With the release of 2.0 Streams will introduce topology optimization. We > should provide a config with a default value of false allowing users to > enable/disable optimization. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8530) Consumer should handle authorization errors in OffsetFetch
[
https://issues.apache.org/jira/browse/KAFKA-8530?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864181#comment-16864181
]
ASF GitHub Bot commented on KAFKA-8530:
---
hachikuji commented on pull request #6928: KAFKA-8530; Check for topic
authorization errors in OffsetFetch response
URL: https://github.com/apache/kafka/pull/6928
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Consumer should handle authorization errors in OffsetFetch
> --
>
> Key: KAFKA-8530
> URL: https://issues.apache.org/jira/browse/KAFKA-8530
> Project: Kafka
> Issue Type: Bug
> Components: consumer
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>Priority: Major
>
> Found this in a test failure:
> {code:java}
> 10:40:43 kafka.api.PlaintextEndToEndAuthorizationTest >
> testNoConsumeWithoutDescribeAclViaSubscribe FAILED
> 10:40:43 java.lang.Exception: Unexpected exception,
> expected but
> was
> 10:40:43
> 10:40:43 Caused by:
> 10:40:43 org.apache.kafka.common.KafkaException: Unexpected error in fetch
> offset response for partition e2etopic-0: Not authorized to access topics:
> [Topic authorization failed.]{code}
> The problem is that we are not checking for authorization errors in the
> offset fetch response. Rather than throwing {{TopicAuthorizationException}},
> we currently raise a plain {{KafkaException}}. The test case works most of
> the time because we usually see the authorization error when the first
> Metadata request is sent. It's unclear why that didn't happen here. Possibly
> there is a race condition setting the initial ACLs.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (KAFKA-8530) Consumer should handle authorization errors in OffsetFetch
[
https://issues.apache.org/jira/browse/KAFKA-8530?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jason Gustafson resolved KAFKA-8530.
Resolution: Fixed
Fix Version/s: 2.4.0
> Consumer should handle authorization errors in OffsetFetch
> --
>
> Key: KAFKA-8530
> URL: https://issues.apache.org/jira/browse/KAFKA-8530
> Project: Kafka
> Issue Type: Bug
> Components: consumer
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>Priority: Major
> Fix For: 2.4.0
>
>
> Found this in a test failure:
> {code:java}
> 10:40:43 kafka.api.PlaintextEndToEndAuthorizationTest >
> testNoConsumeWithoutDescribeAclViaSubscribe FAILED
> 10:40:43 java.lang.Exception: Unexpected exception,
> expected but
> was
> 10:40:43
> 10:40:43 Caused by:
> 10:40:43 org.apache.kafka.common.KafkaException: Unexpected error in fetch
> offset response for partition e2etopic-0: Not authorized to access topics:
> [Topic authorization failed.]{code}
> The problem is that we are not checking for authorization errors in the
> offset fetch response. Rather than throwing {{TopicAuthorizationException}},
> we currently raise a plain {{KafkaException}}. The test case works most of
> the time because we usually see the authorization error when the first
> Metadata request is sent. It's unclear why that didn't happen here. Possibly
> there is a race condition setting the initial ACLs.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-8532) controller-event-thread deadlock with zk-session-expiry-handler0
[ https://issues.apache.org/jira/browse/KAFKA-8532?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864184#comment-16864184 ] Jun Rao commented on KAFKA-8532: [~lbdai3190], expiryScheduler.shutdown() is probably not the issue. That part of the code blocks the KafkaServer startup through initZkClient(). At that point, the controller hasn't been started and therefore, there is no controller event thread yet. > controller-event-thread deadlock with zk-session-expiry-handler0 > > > Key: KAFKA-8532 > URL: https://issues.apache.org/jira/browse/KAFKA-8532 > Project: Kafka > Issue Type: Bug > Components: core >Affects Versions: 2.1.1 >Reporter: leibo >Priority: Blocker > Attachments: js.log > > > We have observed a serious deadlock between controller-event-thead and > zk-session-expirey-handle thread. When this issue occurred, it's only one way > to recovery the kafka cluster is restart kafka server. The follows is the > jstack log of controller-event-thead and zk-session-expiry-handle thread. > "zk-session-expiry-handler0" #163089 daemon prio=5 os_prio=0 > tid=0x7fcc9c01 nid=0xfb22 waiting on condition [0x7fcbb01f8000] > java.lang.Thread.State: WAITING (parking) > at sun.misc.Unsafe.park(Native Method) > - parking to wait for <0x0005ee3f7000> (a > java.util.concurrent.CountDownLatch$Sync) > at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) > at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231) // > 等待controller-event-thread线程处理expireEvent > at > kafka.controller.KafkaController$Expire.waitUntilProcessingStarted(KafkaController.scala:1533) > at > kafka.controller.KafkaController$$anon$7.beforeInitializingSession(KafkaController.scala:173) > at > kafka.zookeeper.ZooKeeperClient.callBeforeInitializingSession(ZooKeeperClient.scala:408) > at > kafka.zookeeper.ZooKeeperClient.$anonfun$reinitialize$1(ZooKeeperClient.scala:374) > at > kafka.zookeeper.ZooKeeperClient.$anonfun$reinitialize$1$adapted(ZooKeeperClient.scala:374) > at kafka.zookeeper.ZooKeeperClient$$Lambda$1473/1823438251.apply(Unknown > Source) > at scala.collection.Iterator.foreach(Iterator.scala:937) > at scala.collection.Iterator.foreach$(Iterator.scala:937) > at scala.collection.AbstractIterator.foreach(Iterator.scala:1425) > at scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:209) > at kafka.zookeeper.ZooKeeperClient.reinitialize(ZooKeeperClient.scala:374) > at > kafka.zookeeper.ZooKeeperClient.$anonfun$scheduleSessionExpiryHandler$1(ZooKeeperClient.scala:428) > at > kafka.zookeeper.ZooKeeperClient$$Lambda$1471/701792920.apply$mcV$sp(Unknown > Source) > at kafka.utils.KafkaScheduler.$anonfun$schedule$2(KafkaScheduler.scala:114) > at kafka.utils.KafkaScheduler$$Lambda$198/1048098469.apply$mcV$sp(Unknown > Source) > at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:63) > at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) > at java.util.concurrent.FutureTask.run(FutureTask.java:266) > at > java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) > at > java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) > at > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > at > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > at java.lang.Thread.run(Thread.java:748) > Locked ownable synchronizers: > - <0x000661e8d2e0> (a java.util.concurrent.ThreadPoolExecutor$Worker) > "controller-event-thread" #51 prio=5 os_prio=0 tid=0x7fceaeec4000 > nid=0x310 waiting on condition [0x7fccb55c8000] > java.lang.Thread.State: WAITING (parking) > at sun.misc.Unsafe.park(Native Method) > - parking to wait for <0x0005d1be5a00> (a > java.util.concurrent.CountDownLatch$Sync) > at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997) > at > java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304) > at ja
[jira] [Commented] (KAFKA-8452) Possible Suppress buffer optimization: de-duplicate prior value
[ https://issues.apache.org/jira/browse/KAFKA-8452?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864194#comment-16864194 ] ASF GitHub Bot commented on KAFKA-8452: --- vvcephei commented on pull request #6940: KAFKA-8452: Compressed BufferValue review follow-up URL: https://github.com/apache/kafka/pull/6940 Belatedly address a few code review comments from #6848 ### Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Possible Suppress buffer optimization: de-duplicate prior value > --- > > Key: KAFKA-8452 > URL: https://issues.apache.org/jira/browse/KAFKA-8452 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Priority: Major > > As of KAFKA-8199, the suppression buffers have to track the "prior value" in > addition to the "old" and "new" values for each record, to support > transparent downstream views. > In many cases, the prior value is actually the same as the old value, and we > could avoid storing it separately. The challenge is that the old and new > values are already serialized into a common array (as a Change via the > FullChangeSerde), so the "prior" value would actually be a slice on the > underlying array. But, of course, Java does not have array slices. > To get around this, we either need to switch to ByteBuffers (which support > slices) or break apart the serialized Change into just serialized old and new > values. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8488) FetchSessionHandler logging create 73 mb allocation in TLAB which could be no op
[ https://issues.apache.org/jira/browse/KAFKA-8488?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864233#comment-16864233 ] ASF GitHub Bot commented on KAFKA-8488: --- cmccabe commented on pull request #6890: KAFKA-8488: guarded debug log to avoid frequent toString allocation URL: https://github.com/apache/kafka/pull/6890 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > FetchSessionHandler logging create 73 mb allocation in TLAB which could be no > op > - > > Key: KAFKA-8488 > URL: https://issues.apache.org/jira/browse/KAFKA-8488 > Project: Kafka > Issue Type: Improvement >Reporter: Wenshuai Hou >Priority: Minor > Attachments: image-2019-06-05-14-04-35-668.png > > > !image-2019-06-05-14-04-35-668.png! -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (KAFKA-8452) Possible Suppress buffer optimization: de-duplicate prior value
[ https://issues.apache.org/jira/browse/KAFKA-8452?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax reassigned KAFKA-8452: -- Assignee: John Roesler > Possible Suppress buffer optimization: de-duplicate prior value > --- > > Key: KAFKA-8452 > URL: https://issues.apache.org/jira/browse/KAFKA-8452 > Project: Kafka > Issue Type: Improvement > Components: streams >Reporter: John Roesler >Assignee: John Roesler >Priority: Major > > As of KAFKA-8199, the suppression buffers have to track the "prior value" in > addition to the "old" and "new" values for each record, to support > transparent downstream views. > In many cases, the prior value is actually the same as the old value, and we > could avoid storing it separately. The challenge is that the old and new > values are already serialized into a common array (as a Change via the > FullChangeSerde), so the "prior" value would actually be a slice on the > underlying array. But, of course, Java does not have array slices. > To get around this, we either need to switch to ByteBuffers (which support > slices) or break apart the serialized Change into just serialized old and new > values. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-8540) Improve InsertField connect transform to allow inserting processed timestamp
Khushal Kumthekar created KAFKA-8540:
Summary: Improve InsertField connect transform to allow inserting
processed timestamp
Key: KAFKA-8540
URL: https://issues.apache.org/jira/browse/KAFKA-8540
Project: Kafka
Issue Type: Improvement
Components: KafkaConnect
Affects Versions: 2.2.1
Reporter: Khushal Kumthekar
We have requirement to insert a new field that gets populated with the
timestamp when kafka message was processed by the connector.
There is no such transform currently available. *"InputField"* transform
provides *timestamp.field* option but that sets the field value to the
timestamp when messaged landed in kafka and not when message was processed by
the connector.
Here is the proposal
Enhance *InputField* transform to add a new option called
*"processedTimestamp.field"* that will allow users to insert new field with
value of message processed timestamp as shown in below configuration
{code:java}
"transforms": "InsertField",
"transforms.InsertField.type":"org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.InsertField.processedTimestamp.field": "field_name"{code}
here is the potential code block to set this value
{code:java}
updatedValue.put(processedTimestampField.name, new Date()); {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-8456) Flaky Test StoreUpgradeIntegrationTest#shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi
[ https://issues.apache.org/jira/browse/KAFKA-8456?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864354#comment-16864354 ] Matthias J. Sax commented on KAFKA-8456: [https://builds.apache.org/blue/organizations/jenkins/kafka-trunk-jdk8/detail/kafka-trunk-jdk8/3723/tests] > Flaky Test > StoreUpgradeIntegrationTest#shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > --- > > Key: KAFKA-8456 > URL: https://issues.apache.org/jira/browse/KAFKA-8456 > Project: Kafka > Issue Type: Bug > Components: streams, unit tests >Reporter: Boyang Chen >Priority: Major > > [https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/22331/console] > *01:20:07* > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > failed, log available in > /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/streams/build/reports/testOutput/org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi.test.stdout*01:20:07* > *01:20:07* org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest > > shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > FAILED*01:20:07* java.lang.AssertionError: Condition not met within > timeout 15000. Could not get expected result in time.*01:20:07* at > org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:375)*01:20:07* > at > org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:335)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.verifyWindowedCountWithTimestamp(StoreUpgradeIntegrationTest.java:830)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(StoreUpgradeIntegrationTest.java:573)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi(StoreUpgradeIntegrationTest.java:517) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8108) Flaky Test kafka.api.ClientIdQuotaTest.testThrottledProducerConsumer
[
https://issues.apache.org/jira/browse/KAFKA-8108?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864353#comment-16864353
]
Matthias J. Sax commented on KAFKA-8108:
[https://builds.apache.org/blue/organizations/jenkins/kafka-trunk-jdk8/detail/kafka-trunk-jdk8/3722/tests]
> Flaky Test kafka.api.ClientIdQuotaTest.testThrottledProducerConsumer
>
>
> Key: KAFKA-8108
> URL: https://issues.apache.org/jira/browse/KAFKA-8108
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 2.3.0
>Reporter: Guozhang Wang
>Priority: Critical
> Labels: flaky-test
> Fix For: 2.4.0
>
>
> {code}
> java.lang.AssertionError: Client with id=QuotasTestProducer-!@#$%^&*() should
> have been throttled
> at org.junit.Assert.fail(Assert.java:89)
> at org.junit.Assert.assertTrue(Assert.java:42)
> at
> kafka.api.QuotaTestClients.verifyThrottleTimeMetric(BaseQuotaTest.scala:229)
> at
> kafka.api.QuotaTestClients.verifyProduceThrottle(BaseQuotaTest.scala:215)
> at
> kafka.api.BaseQuotaTest.testThrottledProducerConsumer(BaseQuotaTest.scala:82)
> {code}
> https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/3230/testReport/junit/kafka.api/ClientIdQuotaTest/testThrottledProducerConsumer/
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Assigned] (KAFKA-8456) Flaky Test StoreUpgradeIntegrationTest#shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi
[ https://issues.apache.org/jira/browse/KAFKA-8456?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax reassigned KAFKA-8456: -- Assignee: Matthias J. Sax > Flaky Test > StoreUpgradeIntegrationTest#shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > --- > > Key: KAFKA-8456 > URL: https://issues.apache.org/jira/browse/KAFKA-8456 > Project: Kafka > Issue Type: Bug > Components: streams, unit tests >Reporter: Boyang Chen >Assignee: Matthias J. Sax >Priority: Major > > [https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/22331/console] > *01:20:07* > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > failed, log available in > /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/streams/build/reports/testOutput/org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi.test.stdout*01:20:07* > *01:20:07* org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest > > shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > FAILED*01:20:07* java.lang.AssertionError: Condition not met within > timeout 15000. Could not get expected result in time.*01:20:07* at > org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:375)*01:20:07* > at > org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:335)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.verifyWindowedCountWithTimestamp(StoreUpgradeIntegrationTest.java:830)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(StoreUpgradeIntegrationTest.java:573)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi(StoreUpgradeIntegrationTest.java:517) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-8541) Flakey test LeaderElectionCommandTest#testTopicPartition
Boyang Chen created KAFKA-8541: -- Summary: Flakey test LeaderElectionCommandTest#testTopicPartition Key: KAFKA-8541 URL: https://issues.apache.org/jira/browse/KAFKA-8541 Project: Kafka Issue Type: Bug Components: core Reporter: Boyang Chen [https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/5528/consoleFull] *01:37:34* kafka.admin.LeaderElectionCommandTest > testTopicPartition STARTED*01:38:13* kafka.admin.LeaderElectionCommandTest.testTopicPartition failed, log available in /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk11-scala2.12/core/build/reports/testOutput/kafka.admin.LeaderElectionCommandTest.testTopicPartition.test.stdout*01:38:13* *01:38:13* kafka.admin.LeaderElectionCommandTest > testTopicPartition FAILED*01:38:13* kafka.common.AdminCommandFailedException: Timeout waiting for election results*01:38:13* at kafka.admin.LeaderElectionCommand$.electLeaders(LeaderElectionCommand.scala:134)*01:38:13* at kafka.admin.LeaderElectionCommand$.run(LeaderElectionCommand.scala:89)*01:38:13* at kafka.admin.LeaderElectionCommand$.main(LeaderElectionCommand.scala:42)*01:38:13* at kafka.admin.LeaderElectionCommandTest.$anonfun$testTopicPartition$1(LeaderElectionCommandTest.scala:125)*01:38:13* at kafka.admin.LeaderElectionCommandTest.$anonfun$testTopicPartition$1$adapted(LeaderElectionCommandTest.scala:103)*01:38:13* at kafka.utils.TestUtils$.resource(TestUtils.scala:1528)*01:38:13* at kafka.admin.LeaderElectionCommandTest.testTopicPartition(LeaderElectionCommandTest.scala:103)*01:38:13* *01:38:13* Caused by:*01:38:13* org.apache.kafka.common.errors.TimeoutException: Aborted due to timeout.*01:38:13* -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Closed] (KAFKA-4048) Connect does not support RetriableException consistently for sinks
[
https://issues.apache.org/jira/browse/KAFKA-4048?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ewen Cheslack-Postava closed KAFKA-4048.
> Connect does not support RetriableException consistently for sinks
> --
>
> Key: KAFKA-4048
> URL: https://issues.apache.org/jira/browse/KAFKA-4048
> Project: Kafka
> Issue Type: Bug
> Components: KafkaConnect
>Reporter: Shikhar Bhushan
>Assignee: Shikhar Bhushan
>Priority: Major
>
> We only allow for handling {{RetriableException}} from calls to
> {{SinkTask.put()}}, but this is something we should support also for
> {{flush()}} and arguably also {{open()}}.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (KAFKA-8456) Flaky Test StoreUpgradeIntegrationTest#shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi
[ https://issues.apache.org/jira/browse/KAFKA-8456?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864423#comment-16864423 ] ASF GitHub Bot commented on KAFKA-8456: --- mjsax commented on pull request #6941: KAFKA-8456: Stabilize flaky StoreUpgradeIntegrationTest URL: https://github.com/apache/kafka/pull/6941 Running those tests locally, most run about 10 seconds. The default timeout is 15 seconds. I think it makes sense to provide more head room on Jenkins. This PR increase the timeout to 60 seconds. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Flaky Test > StoreUpgradeIntegrationTest#shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > --- > > Key: KAFKA-8456 > URL: https://issues.apache.org/jira/browse/KAFKA-8456 > Project: Kafka > Issue Type: Bug > Components: streams, unit tests >Reporter: Boyang Chen >Assignee: Matthias J. Sax >Priority: Major > > [https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/22331/console] > *01:20:07* > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > failed, log available in > /home/jenkins/jenkins-slave/workspace/kafka-pr-jdk8-scala2.11/streams/build/reports/testOutput/org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi.test.stdout*01:20:07* > *01:20:07* org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest > > shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi > FAILED*01:20:07* java.lang.AssertionError: Condition not met within > timeout 15000. Could not get expected result in time.*01:20:07* at > org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:375)*01:20:07* > at > org.apache.kafka.test.TestUtils.waitForCondition(TestUtils.java:335)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.verifyWindowedCountWithTimestamp(StoreUpgradeIntegrationTest.java:830)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigrateWindowStoreToTimestampedWindowStoreUsingPapi(StoreUpgradeIntegrationTest.java:573)*01:20:07* > at > org.apache.kafka.streams.integration.StoreUpgradeIntegrationTest.shouldMigratePersistentWindowStoreToTimestampedWindowStoreUsingPapi(StoreUpgradeIntegrationTest.java:517) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-8494) Refactor Consumer#StickyAssignor and add CooperativeStickyAssignor (part 4)
[ https://issues.apache.org/jira/browse/KAFKA-8494?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Sophie Blee-Goldman updated KAFKA-8494: --- Summary: Refactor Consumer#StickyAssignor and add CooperativeStickyAssignor (part 4) (was: Refactor Consumer#StickyAssignor to support incremental protocol (part 4)) > Refactor Consumer#StickyAssignor and add CooperativeStickyAssignor (part 4) > --- > > Key: KAFKA-8494 > URL: https://issues.apache.org/jira/browse/KAFKA-8494 > Project: Kafka > Issue Type: Sub-task >Reporter: Guozhang Wang >Priority: Major > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8179) Incremental Rebalance Protocol for Kafka Consumer
[ https://issues.apache.org/jira/browse/KAFKA-8179?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864496#comment-16864496 ] ASF GitHub Bot commented on KAFKA-8179: --- guozhangwang commented on pull request #6528: KAFKA-8179: Part I, Bump up consumer protocol to v2 URL: https://github.com/apache/kafka/pull/6528 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org > Incremental Rebalance Protocol for Kafka Consumer > - > > Key: KAFKA-8179 > URL: https://issues.apache.org/jira/browse/KAFKA-8179 > Project: Kafka > Issue Type: Improvement > Components: consumer >Reporter: Guozhang Wang >Assignee: Guozhang Wang >Priority: Major > > Recently Kafka community is promoting cooperative rebalancing to mitigate the > pain points in the stop-the-world rebalancing protocol. This ticket is > created to initiate that idea at the Kafka consumer client, which will be > beneficial for heavy-stateful consumers such as Kafka Streams applications. > In short, the scope of this ticket includes reducing unnecessary rebalance > latency due to heavy partition migration: i.e. partitions being revoked and > re-assigned. This would make the built-in consumer assignors (range, > round-robin etc) to be aware of previously assigned partitions and be sticky > in best-effort. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (KAFKA-8106) Reducing the allocation and copying of ByteBuffer when logValidator do validation.
[
https://issues.apache.org/jira/browse/KAFKA-8106?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16864532#comment-16864532
]
Guozhang Wang commented on KAFKA-8106:
--
Okay guys after some further benchmarks now I think I've finally realized the
difference in perf:
1. our current code would allocate a byte buffer for each record deserialized:
{code}
ByteBuffer recordBuffer = ByteBuffer.allocate(sizeOfBodyInBytes);
input.readFully(recordBuffer.array(), 0, sizeOfBodyInBytes);
{code}
the byte array size is `sizeOfBodyInBytes`, and it can cause GC pressure;
2. the skipBytes implementation tries to not allocate such a big byte array;
HOWEVER depending on the compressionType the underlying implementation is
different:
2.a. LZ4 used KafkaLZ4BlockInputStream, which has a shared decompressionBuffer,
default size 65536. It means, for a batch of records, we will only have a
single allocated buffer.
2.b. All other compressionType used BufferedDataInputStream, whose skipBytes is
implemented as
{code}
int size = (int)Math.min(MAX_SKIP_BUFFER_SIZE, remaining);
byte[] skipBuffer = new byte[size];
while (remaining > 0) {
nr = read(skipBuffer, 0, (int)Math.min(size, remaining));
if (nr < 0) {
break;
}
remaining -= nr;
}
{code}
I.e. each call will allocate a new byte buffer, which is different from 2.a.
3. Our current implementation is slightly better than 2.b because we still only
allocate one buffer although we need to `skip` three fields (key, value, and
array of headers). But it is still inferior to 2.a, which only use a single
buffer; current implementation is ONLY beneficial if the record size is larger
than 2048, whereas 2.a's approach, as demonstrated by the original author whose
used LZ4, is much better even for 1KB message size. So our perf numbers on
other compression types than LZ4 would not show much benefits until record size
is much larger than 2048.
So I think if we want to get the similar performance boost for all compression
types as the original PR did for LZ4, we then need to have single shared buffer
associated with the `InputStream` object, generated from `wrapForInput`, which
can then be used for all the records within a batch. (edited)
> Reducing the allocation and copying of ByteBuffer when logValidator do
> validation.
>
>
> Key: KAFKA-8106
> URL: https://issues.apache.org/jira/browse/KAFKA-8106
> Project: Kafka
> Issue Type: Bug
> Components: core
>Affects Versions: 2.2.0, 2.1.1
> Environment: Server :
> cpu:2*16 ;
> MemTotal : 256G;
> Ethernet controller:Intel Corporation 82599ES 10-Gigabit SFI/SFP+ Network
> Connection ;
> SSD.
>Reporter: Flower.min
>Assignee: Flower.min
>Priority: Major
> Labels: performance
>
> We do performance testing about Kafka in specific scenarios as
> described below .We build a kafka cluster with one broker,and create topics
> with different number of partitions.Then we start lots of producer processes
> to send large amounts of messages to one of the topics at one testing .
> *_Specific Scenario_*
>
> *_1.Main config of Kafka_*
> # Main config of Kafka
> server:[num.network.threads=6;num.io.threads=128;queued.max.requests=500|http://num.network.threads%3D6%3Bnum.io.threads%3D128%3Bqueued.max.requests%3D500/]
> # Number of TopicPartition : 50~2000
> # Size of Single Message : 1024B
>
> *_2.Config of KafkaProducer_*
> ||compression.type||[linger.ms|http://linger.ms/]||batch.size||buffer.memory||
> |lz4|1000ms~5000ms|16KB/10KB/100KB|128MB|
> *_3.The best result of performance testing_*
> ||Network inflow rate||CPU Used (%)||Disk write speed||Performance of
> production||
> |550MB/s~610MB/s|97%～99%|550MB/s~610MB/s |23,000,000 messages/s|
> *_4.Phenomenon and my doubt_*
> _The upper limit of CPU usage has been reached But it does not
> reach the upper limit of the bandwidth of the server network. *We are
> doubtful about which cost too much CPU time and we want to Improve
> performance and reduces CPU usage of Kafka server.*_
> _*5.Analysis*_
> We analysis the JFIR of Kafka server when doing performance testing
> .After we checked and completed the performance test again, we located the
> code "*ByteBuffer recordBuffer =
> ByteBuffer.allocate(sizeOfBodyInBytes);*(*Class:DefaultRecord,Function:readFrom()*)”
> which consumed CPU resources and caused a lot of GC .Our modified code
> reduces the allocation and copying of ByteBuffer, so the test performance is
> greatly improved, and the CPU's stable usage is *below 60%*. The following is
> a comparison of different code test performance
[jira] [Updated] (KAFKA-8510) Update StreamsPartitionAssignor to use the built-in owned partitions to achieve stickiness (part 7)
[ https://issues.apache.org/jira/browse/KAFKA-8510?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Boyang Chen updated KAFKA-8510: --- Component/s: consumer > Update StreamsPartitionAssignor to use the built-in owned partitions to > achieve stickiness (part 7) > --- > > Key: KAFKA-8510 > URL: https://issues.apache.org/jira/browse/KAFKA-8510 > Project: Kafka > Issue Type: Sub-task > Components: consumer, streams >Reporter: Guozhang Wang >Priority: Major > > Today this information is encoded as part of the user data bytes, we can now > remove it and leverage on the owned partitions of the protocol directly. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-3333) Alternative Partitioner to Support "Always Round-Robin" partitioning
[ https://issues.apache.org/jira/browse/KAFKA-?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-: --- Labels: kip (was: needs-kip patch) > Alternative Partitioner to Support "Always Round-Robin" partitioning > > > Key: KAFKA- > URL: https://issues.apache.org/jira/browse/KAFKA- > Project: Kafka > Issue Type: New Feature > Components: clients >Reporter: Stephen Powis >Assignee: M. Manna >Priority: Major > Labels: kip > Fix For: 2.4.0 > > Original Estimate: 1h > Remaining Estimate: 1h > > ** Please Look into KAFKA-7358 for the official description ** > > The > [DefaultPartitioner|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java] > typically distributes using the hash of the keybytes, and falls back to > round robin if there is no key. But there is currently no way to do Round > Robin partitioning if you have keys on your messages without writing your own > partitioning implementation. > I think it'd be helpful to have an implementation of straight Round Robin > partitioning included with the library. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-3333) Alternative Partitioner to Support "Always Round-Robin" partitioning
[ https://issues.apache.org/jira/browse/KAFKA-?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-: --- Description: KIP: [https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=89070828] Please Look into KAFKA-7358 for the official description ** The [DefaultPartitioner|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java] typically distributes using the hash of the keybytes, and falls back to round robin if there is no key. But there is currently no way to do Round Robin partitioning if you have keys on your messages without writing your own partitioning implementation. I think it'd be helpful to have an implementation of straight Round Robin partitioning included with the library. was: ** Please Look into KAFKA-7358 for the official description ** The [DefaultPartitioner|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java] typically distributes using the hash of the keybytes, and falls back to round robin if there is no key. But there is currently no way to do Round Robin partitioning if you have keys on your messages without writing your own partitioning implementation. I think it'd be helpful to have an implementation of straight Round Robin partitioning included with the library. > Alternative Partitioner to Support "Always Round-Robin" partitioning > > > Key: KAFKA- > URL: https://issues.apache.org/jira/browse/KAFKA- > Project: Kafka > Issue Type: New Feature > Components: clients >Reporter: Stephen Powis >Assignee: M. Manna >Priority: Major > Labels: kip > Fix For: 2.4.0 > > Original Estimate: 1h > Remaining Estimate: 1h > > KIP: > [https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=89070828] > Please Look into KAFKA-7358 for the official description ** > The > [DefaultPartitioner|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java] > typically distributes using the hash of the keybytes, and falls back to > round robin if there is no key. But there is currently no way to do Round > Robin partitioning if you have keys on your messages without writing your own > partitioning implementation. > I think it'd be helpful to have an implementation of straight Round Robin > partitioning included with the library. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
