[jira] [Commented] (KAFKA-2084) byte rate metrics per client ID (producer and consumer)
[ https://issues.apache.org/jira/browse/KAFKA-2084?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14681248#comment-14681248 ] Aditya A Auradkar commented on KAFKA-2084: -- Updated reviewboard https://reviews.apache.org/r/33049/diff/ against branch origin/trunk byte rate metrics per client ID (producer and consumer) --- Key: KAFKA-2084 URL: https://issues.apache.org/jira/browse/KAFKA-2084 Project: Kafka Issue Type: Sub-task Reporter: Aditya Auradkar Assignee: Aditya Auradkar Labels: quotas Attachments: KAFKA-2084.patch, KAFKA-2084_2015-04-09_18:10:56.patch, KAFKA-2084_2015-04-10_17:24:34.patch, KAFKA-2084_2015-04-21_12:21:18.patch, KAFKA-2084_2015-04-21_12:28:05.patch, KAFKA-2084_2015-05-05_15:27:35.patch, KAFKA-2084_2015-05-05_17:52:02.patch, KAFKA-2084_2015-05-11_16:16:01.patch, KAFKA-2084_2015-05-26_11:50:50.patch, KAFKA-2084_2015-06-02_17:02:00.patch, KAFKA-2084_2015-06-02_17:09:28.patch, KAFKA-2084_2015-06-02_17:10:52.patch, KAFKA-2084_2015-06-04_16:31:22.patch, KAFKA-2084_2015-06-12_10:39:35.patch, KAFKA-2084_2015-06-29_17:53:44.patch, KAFKA-2084_2015-08-04_18:50:51.patch, KAFKA-2084_2015-08-04_19:07:46.patch, KAFKA-2084_2015-08-07_11:27:51.patch, KAFKA-2084_2015-08-10_13:48:50.patch, KAFKA-2084_2015-08-10_21:57:48.patch We need to be able to track the bytes-in/bytes-out rate on a per-client ID basis. This is necessary for quotas. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 33049: Patch for KAFKA-2084
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 11, 2015, 4:58 a.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description (updated) --- Updated patch for quotas. This patch does the following: 1. Add per-client metrics for both producer and consumers 2. Add configuration for quotas 3. Compute delay times in the metrics package and return the delay times in QuotaViolationException 4. Add a DelayQueue in KafkaApi's that can be used to throttle any type of request. Implemented request throttling for produce and fetch requests. 5. Added unit and integration test cases for both producer and consumer 6. This doesn't include a system test. There is a separate ticket for that 7. Fixed KAFKA-2191 - (Included fix from : https://reviews.apache.org/r/34418/ ) Addressed comments from Joel and Jun Diffs - clients/src/main/java/org/apache/kafka/common/metrics/Quota.java d82bb0c055e631425bc1ebbc7d387baac76aeeaa clients/src/main/java/org/apache/kafka/common/metrics/QuotaViolationException.java a451e5385c9eca76b38b425e8ac856b2715fcffe clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java ca823fd4639523018311b814fde69b6177e73b97 clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java 98429da34418f7f1deba1b5e44e2e6025212edb3 clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java 544e120594de78c43581a980b1e4087b4fb98ccb clients/src/test/java/org/apache/kafka/common/utils/MockTime.java core/src/main/scala/kafka/server/ClientQuotaManager.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala 7ea509c2c41acc00430c74e025e069a833aac4e7 core/src/main/scala/kafka/server/KafkaConfig.scala dbe170f87331f43e2dc30165080d2cb7dfe5fdbf core/src/main/scala/kafka/server/KafkaServer.scala 84d4730ac634f9a5bf12a656e422fea03ad72da8 core/src/main/scala/kafka/server/ReplicaManager.scala 795220e7f63d163be90738b4c1a39687b44c1395 core/src/main/scala/kafka/server/ThrottledResponse.scala PRE-CREATION core/src/main/scala/kafka/utils/ShutdownableThread.scala fc226c863095b7761290292cd8755cd7ad0f155c core/src/test/scala/integration/kafka/api/QuotasTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala f32d206d3f52f3f9f4d649c213edd7058f4b6150 core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/33049/diff/ Testing --- Thanks, Aditya Auradkar
Re: Review Request 33049: Patch for KAFKA-2084
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 11, 2015, 4:57 a.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description (updated) --- Signed-off-by: Aditya Auradkar aaurad...@linkedin.com Addressing Joel's comments Minor imports changes Added testcase to verify that replication traffic is not throttled Tmp commit Fixing test failure Minor Addressing Joel's comments Addressing comments Addressing comments Addressing Juns comments Minor checkstyle changes fixed test case Addressing Juns comments Diffs (updated) - clients/src/main/java/org/apache/kafka/common/metrics/Quota.java d82bb0c055e631425bc1ebbc7d387baac76aeeaa clients/src/main/java/org/apache/kafka/common/metrics/QuotaViolationException.java a451e5385c9eca76b38b425e8ac856b2715fcffe clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java ca823fd4639523018311b814fde69b6177e73b97 clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java 98429da34418f7f1deba1b5e44e2e6025212edb3 clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java 544e120594de78c43581a980b1e4087b4fb98ccb clients/src/test/java/org/apache/kafka/common/utils/MockTime.java core/src/main/scala/kafka/server/ClientQuotaManager.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala 7ea509c2c41acc00430c74e025e069a833aac4e7 core/src/main/scala/kafka/server/KafkaConfig.scala dbe170f87331f43e2dc30165080d2cb7dfe5fdbf core/src/main/scala/kafka/server/KafkaServer.scala 84d4730ac634f9a5bf12a656e422fea03ad72da8 core/src/main/scala/kafka/server/ReplicaManager.scala 795220e7f63d163be90738b4c1a39687b44c1395 core/src/main/scala/kafka/server/ThrottledResponse.scala PRE-CREATION core/src/main/scala/kafka/utils/ShutdownableThread.scala fc226c863095b7761290292cd8755cd7ad0f155c core/src/test/scala/integration/kafka/api/QuotasTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala f32d206d3f52f3f9f4d649c213edd7058f4b6150 core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/33049/diff/ Testing --- Thanks, Aditya Auradkar
Re: Review Request 33378: Patch for KAFKA-2136
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33378/#review94876 --- Jun/Joel - Thanks for the comments. I'd like to address these all at once after KAFKA-2084 is committed because I will need to rebase after that. - Aditya Auradkar On July 13, 2015, 8:36 p.m., Aditya Auradkar wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33378/ --- (Updated July 13, 2015, 8:36 p.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2136 https://issues.apache.org/jira/browse/KAFKA-2136 Repository: kafka Description --- Changes are - Addressing Joel's comments - protocol changes to the fetch request and response to return the throttle_time_ms to clients - New producer/consumer metrics to expose the avg and max delay time for a client - Test cases. - Addressed Joel's comments For now the patch will publish a zero delay and return a response Diffs - clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java 56281ee15cc33dfc96ff64d5b1e596047c7132a4 clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java 07e65d4a54ba4eef5b787eba3e71cbe9f6a920bd clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java 8686d83aa52e435c6adafbe9ff4bd1602281072a clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java eb8951fba48c335095cc43fc3672de1c733e07ff clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java fabeae3083a8ea55cdacbb9568f3847ccd85bab4 clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java 37ec0b79beafcf5735c386b066eb319fb697eff5 clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java 419541011d652becf0cda7a5e62ce813cddb1732 clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java 8b1805d3d2bcb9fe2bacb37d870c3236aa9532c4 clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java e3cc1967e407b64cc734548c19e30de700b64ba8 core/src/main/scala/kafka/api/FetchRequest.scala 5b38f8554898e54800abd65a7415dd0ac41fd958 core/src/main/scala/kafka/api/FetchResponse.scala 0b6b33ab6f7a732ff1322b6f48abd4c43e0d7215 core/src/main/scala/kafka/api/ProducerRequest.scala c866180d3680da03e48d374415f10220f6ca68c4 core/src/main/scala/kafka/api/ProducerResponse.scala 5d1fac4cb8943f5bfaa487f8e9d9d2856efbd330 core/src/main/scala/kafka/consumer/SimpleConsumer.scala c16f7edd322709060e54c77eb505c44cbd77a4ec core/src/main/scala/kafka/server/AbstractFetcherThread.scala 83fc47417dd7fe4edf030217fa7fd69d99b170b0 core/src/main/scala/kafka/server/DelayedFetch.scala de6cf5bdaa0e70394162febc63b50b55ca0a92db core/src/main/scala/kafka/server/DelayedProduce.scala 05078b24ef28f2f4e099afa943e43f1d00359fda core/src/main/scala/kafka/server/KafkaApis.scala d63bc18d795a6f0e6994538ca55a9a46f7fb8ffd core/src/main/scala/kafka/server/OffsetManager.scala 5cca85cf727975f6d3acb2223fd186753ad761dc core/src/main/scala/kafka/server/ReplicaFetcherThread.scala b31b432a226ba79546dd22ef1d2acbb439c2e9a3 core/src/main/scala/kafka/server/ReplicaManager.scala 59c9bc3ac3a8afc07a6f8c88c5871304db588d17 core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala 5717165f2344823fabe8f7cfafae4bb8af2d949a core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala 00d59337a99ac135e8689bd1ecd928f7b1423d79 Diff: https://reviews.apache.org/r/33378/diff/ Testing --- New tests added Thanks, Aditya Auradkar
[jira] [Updated] (KAFKA-2084) byte rate metrics per client ID (producer and consumer)
[ https://issues.apache.org/jira/browse/KAFKA-2084?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aditya A Auradkar updated KAFKA-2084: - Attachment: KAFKA-2084_2015-08-10_21:57:48.patch byte rate metrics per client ID (producer and consumer) --- Key: KAFKA-2084 URL: https://issues.apache.org/jira/browse/KAFKA-2084 Project: Kafka Issue Type: Sub-task Reporter: Aditya Auradkar Assignee: Aditya Auradkar Labels: quotas Attachments: KAFKA-2084.patch, KAFKA-2084_2015-04-09_18:10:56.patch, KAFKA-2084_2015-04-10_17:24:34.patch, KAFKA-2084_2015-04-21_12:21:18.patch, KAFKA-2084_2015-04-21_12:28:05.patch, KAFKA-2084_2015-05-05_15:27:35.patch, KAFKA-2084_2015-05-05_17:52:02.patch, KAFKA-2084_2015-05-11_16:16:01.patch, KAFKA-2084_2015-05-26_11:50:50.patch, KAFKA-2084_2015-06-02_17:02:00.patch, KAFKA-2084_2015-06-02_17:09:28.patch, KAFKA-2084_2015-06-02_17:10:52.patch, KAFKA-2084_2015-06-04_16:31:22.patch, KAFKA-2084_2015-06-12_10:39:35.patch, KAFKA-2084_2015-06-29_17:53:44.patch, KAFKA-2084_2015-08-04_18:50:51.patch, KAFKA-2084_2015-08-04_19:07:46.patch, KAFKA-2084_2015-08-07_11:27:51.patch, KAFKA-2084_2015-08-10_13:48:50.patch, KAFKA-2084_2015-08-10_21:57:48.patch We need to be able to track the bytes-in/bytes-out rate on a per-client ID basis. This is necessary for quotas. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2071) Replace Produce Request/Response with their org.apache.kafka.common.requests equivalents
[ https://issues.apache.org/jira/browse/KAFKA-2071?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14681274#comment-14681274 ] David Jacot commented on KAFKA-2071: I just updated the PR. Now, ProducerRequest and ProducerReponse are completely removed. Regarding the producer, it has been updated but some code duplications have been introduced to support other requests/responses. If I'm not mistaken, TopicMetadata is the only one left in the producer so it would be good to migrate it. I'll check this. Replace Produce Request/Response with their org.apache.kafka.common.requests equivalents Key: KAFKA-2071 URL: https://issues.apache.org/jira/browse/KAFKA-2071 Project: Kafka Issue Type: Sub-task Reporter: Gwen Shapira Assignee: David Jacot Fix For: 0.8.3 Replace Produce Request/Response with their org.apache.kafka.common.requests equivalents -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-2071) Replace Produce Request/Response with their org.apache.kafka.common.requests equivalents
[ https://issues.apache.org/jira/browse/KAFKA-2071?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] David Jacot updated KAFKA-2071: --- Status: Patch Available (was: In Progress) Replace Produce Request/Response with their org.apache.kafka.common.requests equivalents Key: KAFKA-2071 URL: https://issues.apache.org/jira/browse/KAFKA-2071 Project: Kafka Issue Type: Sub-task Reporter: Gwen Shapira Assignee: David Jacot Fix For: 0.8.3 Replace Produce Request/Response with their org.apache.kafka.common.requests equivalents -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2196) remove roundrobin identical topic constraint in consumer coordinator
[ https://issues.apache.org/jira/browse/KAFKA-2196?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680204#comment-14680204 ] Andrew Olson commented on KAFKA-2196: - Doesn't this [1] code also need to be updated? [1] https://github.com/apache/kafka/blob/0.8.2.1/core/src/main/scala/kafka/consumer/PartitionAssignor.scala#L78-82 remove roundrobin identical topic constraint in consumer coordinator Key: KAFKA-2196 URL: https://issues.apache.org/jira/browse/KAFKA-2196 Project: Kafka Issue Type: Sub-task Reporter: Onur Karaman Assignee: Onur Karaman Attachments: KAFKA-2196.patch roundrobin doesn't need to make all consumers have identical topic subscriptions. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 34766: Patch for KAFKA-2229
On Aug. 10, 2015, 3:47 p.m., Grant Henke wrote: My apologies for not looking at this sooner, or suggesting this sooner. Given that this code change and scope is fairly large, would it be too much work to break out the patches reviews by each new protocol message? Then reviews can be more accessable, focus on one thing at a time, and the simple ones can get done much quicker. I am thinking starting with CreateTopic, then DeleteTopic, then AlterTopic may makes sense. If you disagree feel free to say so and I will review this patch as is. Andrii Biletskyi wrote: Hey Grant. Not sure this will work well. First of all this patch is already a part of one bigger feature KIP-4 (which we decided to split into patches :)). Secondly and most importantly, logic for handling all 3 requests relies on some base code. This means we will have to extract that common code (exceptions, some utils other stuff) first and probably submit it as a separate patch, which I'd prefer not to do. But I'm very open to suggestions if it speeds up the review process. Grant Henke wrote: Thanks Andrii, thats understandable. I will pull down and start reviewing. Since this patch is over a month old, it does not look like it applies cleanly to trunk. Can you please update it so it applies cleanly? Andrii, you may also consider submitting a GitHub PR with the rebased branch, see https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes for the instructions. You can stick with Review Board if you prefer it though. Both approaches are still supported. - Ismael --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94749 --- On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala 252ac813c8df1780c2dc5fa9e698fb43bb6d5cf8 core/src/test/scala/unit/kafka/server/TopicCommandHelperTest.scala PRE-CREATION Diff:
Re: [DISCUSSION] KIP-29 - Add an IsrPropagateIntervalMs config to KafkaConfig
Hey Guys, Looks like Jun and Jiangjie have covered the motivation behind the config. To me one would want to set ISR propagation delay to a high number primarily during rolling upgrade. I think the delay one would want to have in ISR propagation is proportional to the cluster size. However, during normal operations a faster ISR propagation is desired. Having a config to expose the delay time provides admin a way to control it, and it will come with a default value so if someone does not want to play with it they can choose not to. @Gwen, does that answer your question? On Sun, Aug 9, 2015 at 3:26 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Jun, Thanks for the detailed explanation. Now I understand. Yes, we might not be able leverage the group commit in this case. When I was testing the patch, I also found a potential use case for the config (not sure if that is a strong use case though). When we rolling upgrade a cluster, if the controller is still running on an old version, the brokers got bounced will create a bunch of ZK path that will not be picked up by the controller until the controller is upgraded to the new version. It might be fine for small clusters, for larger clusters, there could be many ISR change got accumulated before the controller runs on the new version. So if we have the config, when people do rolling upgrade, they can first have the propagation interval to be very large so ISR change will not be propagated. After all the brokers are running on the new version, we can set the propagation interval back to a short value and bounce brokers again. Thanks, Jiangjie (Becket) Qin On Sun, Aug 9, 2015 at 12:22 PM, Jun Rao j...@confluent.io wrote: Jiangjie, Related to group commit, I think what Jay says is the following. Suppose the propagation of ISR is slow and also creates back pressure, you can just propagate the ISR one batch at a time as fast as you can. After one batch is sent, you just collect all ISR changes that have accumulated since the last propagation and propagate them as a batch. This way, the amount of batching happens automatically based on the propagation rate and the delay of propagation is minimized. However, this doesn't quite work if we propagate the ISR changes by writing a notification path in ZK. There is no back pressure: the notification can be written faster than the listener can process it. So the explicit batching that you did essentially is to get around this problem. As for the config, I can see a couple of use cases: (1) We have unit tests that verify the propagation of ISR. To prevent those tests taking too long, we can configure a smaller propagation interval. (2) When there are many partitions and brokers, someone may want to increase the propagation interval to reduce the ZK overhead. I agree that most people likely don't need to change this setting since the default should be good enough. Thanks, Jun On Fri, Aug 7, 2015 at 4:06 PM, Gwen Shapira g...@confluent.io wrote: Maybe Ashish can supply the use-case and tuning advice then :) I'm a -1 on adding new configurations that we can't quite explain. On Fri, Aug 7, 2015 at 3:57 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Hi Gwen, Completely agree with you. I originally just hard coded it to be 10 seconds. Ashish raised this requirement in KAFKA-2406 because people might want to ISR changes get propagated quicker. I don't have a good use case myself. Personally I think hard code it is fine although I don't object to make it configurable. Thanks, Jiangjie (Becket) Qin -- Regards, Ashish
Re: [DISCUSSION] KIP-29 - Add an IsrPropagateIntervalMs config to KafkaConfig
I guess the question, which I think is what Gwen was getting at, is if, rather than making this configurable, it might be possible to make this just work reliably and with the lowest possible latency in some automatic fashion? I raised group commit because that is a way to automatically batch under load. If that doesn't work perhaps there is another way? The challenge as we've seen with obscure configs is that 99% of people can't figure out how to set them so for 99% of people this won't really help them. -Jay On Mon, Aug 10, 2015 at 9:12 AM, Ashish Singh asi...@cloudera.com wrote: Hey Guys, Looks like Jun and Jiangjie have covered the motivation behind the config. To me one would want to set ISR propagation delay to a high number primarily during rolling upgrade. I think the delay one would want to have in ISR propagation is proportional to the cluster size. However, during normal operations a faster ISR propagation is desired. Having a config to expose the delay time provides admin a way to control it, and it will come with a default value so if someone does not want to play with it they can choose not to. @Gwen, does that answer your question? On Sun, Aug 9, 2015 at 3:26 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Jun, Thanks for the detailed explanation. Now I understand. Yes, we might not be able leverage the group commit in this case. When I was testing the patch, I also found a potential use case for the config (not sure if that is a strong use case though). When we rolling upgrade a cluster, if the controller is still running on an old version, the brokers got bounced will create a bunch of ZK path that will not be picked up by the controller until the controller is upgraded to the new version. It might be fine for small clusters, for larger clusters, there could be many ISR change got accumulated before the controller runs on the new version. So if we have the config, when people do rolling upgrade, they can first have the propagation interval to be very large so ISR change will not be propagated. After all the brokers are running on the new version, we can set the propagation interval back to a short value and bounce brokers again. Thanks, Jiangjie (Becket) Qin On Sun, Aug 9, 2015 at 12:22 PM, Jun Rao j...@confluent.io wrote: Jiangjie, Related to group commit, I think what Jay says is the following. Suppose the propagation of ISR is slow and also creates back pressure, you can just propagate the ISR one batch at a time as fast as you can. After one batch is sent, you just collect all ISR changes that have accumulated since the last propagation and propagate them as a batch. This way, the amount of batching happens automatically based on the propagation rate and the delay of propagation is minimized. However, this doesn't quite work if we propagate the ISR changes by writing a notification path in ZK. There is no back pressure: the notification can be written faster than the listener can process it. So the explicit batching that you did essentially is to get around this problem. As for the config, I can see a couple of use cases: (1) We have unit tests that verify the propagation of ISR. To prevent those tests taking too long, we can configure a smaller propagation interval. (2) When there are many partitions and brokers, someone may want to increase the propagation interval to reduce the ZK overhead. I agree that most people likely don't need to change this setting since the default should be good enough. Thanks, Jun On Fri, Aug 7, 2015 at 4:06 PM, Gwen Shapira g...@confluent.io wrote: Maybe Ashish can supply the use-case and tuning advice then :) I'm a -1 on adding new configurations that we can't quite explain. On Fri, Aug 7, 2015 at 3:57 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Hi Gwen, Completely agree with you. I originally just hard coded it to be 10 seconds. Ashish raised this requirement in KAFKA-2406 because people might want to ISR changes get propagated quicker. I don't have a good use case myself. Personally I think hard code it is fine although I don't object to make it configurable. Thanks, Jiangjie (Becket) Qin -- Regards, Ashish
Re: Review Request 34766: Patch for KAFKA-2229
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94749 --- My apologies for not looking at this sooner, or suggesting this sooner. Given that this code change and scope is fairly large, would it be too much work to break out the patches reviews by each new protocol message? Then reviews can be more accessable, focus on one thing at a time, and the simple ones can get done much quicker. I am thinking starting with CreateTopic, then DeleteTopic, then AlterTopic may makes sense. If you disagree feel free to say so and I will review this patch as is. - Grant Henke On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala 252ac813c8df1780c2dc5fa9e698fb43bb6d5cf8 core/src/test/scala/unit/kafka/server/TopicCommandHelperTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/34766/diff/ Testing --- Thanks, Andrii Biletskyi
Re: Review Request 34766: Patch for KAFKA-2229
On Aug. 10, 2015, 3:47 p.m., Grant Henke wrote: My apologies for not looking at this sooner, or suggesting this sooner. Given that this code change and scope is fairly large, would it be too much work to break out the patches reviews by each new protocol message? Then reviews can be more accessable, focus on one thing at a time, and the simple ones can get done much quicker. I am thinking starting with CreateTopic, then DeleteTopic, then AlterTopic may makes sense. If you disagree feel free to say so and I will review this patch as is. Hey Grant. Not sure this will work well. First of all this patch is already a part of one bigger feature KIP-4 (which we decided to split into patches :)). Secondly and most importantly, logic for handling all 3 requests relies on some base code. This means we will have to extract that common code (exceptions, some utils other stuff) first and probably submit it as a separate patch, which I'd prefer not to do. But I'm very open to suggestions if it speeds up the review process. - Andrii --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94749 --- On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala 252ac813c8df1780c2dc5fa9e698fb43bb6d5cf8 core/src/test/scala/unit/kafka/server/TopicCommandHelperTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/34766/diff/ Testing --- Thanks, Andrii Biletskyi
Re: Review Request 34766: Patch for KAFKA-2229
On Aug. 10, 2015, 3:47 p.m., Grant Henke wrote: My apologies for not looking at this sooner, or suggesting this sooner. Given that this code change and scope is fairly large, would it be too much work to break out the patches reviews by each new protocol message? Then reviews can be more accessable, focus on one thing at a time, and the simple ones can get done much quicker. I am thinking starting with CreateTopic, then DeleteTopic, then AlterTopic may makes sense. If you disagree feel free to say so and I will review this patch as is. Andrii Biletskyi wrote: Hey Grant. Not sure this will work well. First of all this patch is already a part of one bigger feature KIP-4 (which we decided to split into patches :)). Secondly and most importantly, logic for handling all 3 requests relies on some base code. This means we will have to extract that common code (exceptions, some utils other stuff) first and probably submit it as a separate patch, which I'd prefer not to do. But I'm very open to suggestions if it speeds up the review process. Thanks Andrii, thats understandable. I will pull down and start reviewing. Since this patch is over a month old, it does not look like it applies cleanly to trunk. Can you please update it so it applies cleanly? - Grant --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94749 --- On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala 252ac813c8df1780c2dc5fa9e698fb43bb6d5cf8 core/src/test/scala/unit/kafka/server/TopicCommandHelperTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/34766/diff/ Testing --- Thanks, Andrii Biletskyi
Re: [jira] [Commented] (KAFKA-1690) new java producer needs ssl support as a client
Thanks for testing out Rajini. I did ran that test in a loop and it never hanged for me. I am hoping you are using the latest patch since the data left over issue is addressed in latest patch. Also if thats an issue SSLConsumerTest and SSLProducerTest will hang too. Did you notice those are having any issues? I am addressing left over reviews will be sending a new patch in a day or two. Thanks, Harsha On August 10, 2015 at 3:35:34 AM, Rajini Sivaram (rajinisiva...@googlemail.com) wrote: I was running a Kafka cluster with the latest SSL patch over the weekend with IBM JRE, and it has been running fine without any issues. There was light load on the cluster throughout and intermittent heavy load, all using SSL clients. However I am seeing an intermittent unit test hang in org.apache.kafka.common.network.SSLSelectorTest.testRenegotiate(). I am not sure if that is related to the IBM JRE I am using for the build. It looks like data left in appReadBuffer after a handshake may not get processed if no more data arrives on the network, causing the test to loop forever. It will be good if this can be fixed (or at least the test commented out) before the code is committed to avoid breaking the build. Even though there are quite a few outstanding review comments, I do agree that being such a big patch, it will be good to commit the code soon and work on minor issues afterwards. On Mon, Aug 10, 2015 at 12:19 AM, Jun Rao (JIRA) j...@apache.org wrote: [ https://issues.apache.org/jira/browse/KAFKA-1690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14679397#comment-14679397 ] Jun Rao commented on KAFKA-1690: [~rsivaram], this jira is getting pretty close to be committed. Could you test this out on an IBM jvm to see if there is any issue especially with respect to the usage of the sendfile api? new java producer needs ssl support as a client --- Key: KAFKA-1690 URL: https://issues.apache.org/jira/browse/KAFKA-1690 Project: Kafka Issue Type: Sub-task Reporter: Joe Stein Assignee: Sriharsha Chintalapani Fix For: 0.8.3 Attachments: KAFKA-1690.patch, KAFKA-1690.patch, KAFKA-1690_2015-05-10_23:20:30.patch, KAFKA-1690_2015-05-10_23:31:42.patch, KAFKA-1690_2015-05-11_16:09:36.patch, KAFKA-1690_2015-05-12_16:20:08.patch, KAFKA-1690_2015-05-15_07:18:21.patch, KAFKA-1690_2015-05-20_14:54:35.patch, KAFKA-1690_2015-05-21_10:37:08.patch, KAFKA-1690_2015-06-03_18:52:29.patch, KAFKA-1690_2015-06-23_13:18:20.patch, KAFKA-1690_2015-07-20_06:10:42.patch, KAFKA-1690_2015-07-20_11:59:57.patch, KAFKA-1690_2015-07-25_12:10:55.patch -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 34766: Patch for KAFKA-2229
On Aug. 10, 2015, 3:47 p.m., Grant Henke wrote: My apologies for not looking at this sooner, or suggesting this sooner. Given that this code change and scope is fairly large, would it be too much work to break out the patches reviews by each new protocol message? Then reviews can be more accessable, focus on one thing at a time, and the simple ones can get done much quicker. I am thinking starting with CreateTopic, then DeleteTopic, then AlterTopic may makes sense. If you disagree feel free to say so and I will review this patch as is. Andrii Biletskyi wrote: Hey Grant. Not sure this will work well. First of all this patch is already a part of one bigger feature KIP-4 (which we decided to split into patches :)). Secondly and most importantly, logic for handling all 3 requests relies on some base code. This means we will have to extract that common code (exceptions, some utils other stuff) first and probably submit it as a separate patch, which I'd prefer not to do. But I'm very open to suggestions if it speeds up the review process. Grant Henke wrote: Thanks Andrii, thats understandable. I will pull down and start reviewing. Since this patch is over a month old, it does not look like it applies cleanly to trunk. Can you please update it so it applies cleanly? Ismael Juma wrote: Andrii, you may also consider submitting a GitHub PR with the rebased branch, see https://cwiki.apache.org/confluence/display/KAFKA/Contributing+Code+Changes for the instructions. You can stick with Review Board if you prefer it though. Both approaches are still supported. @Grant, ok, I will rebase and submit pull request tomorrow. @Ismael, thanks for the suggestion, I will try this out. - Andrii --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94749 --- On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala
[jira] [Assigned] (KAFKA-1811) ensuring registered broker host:port is unique
[ https://issues.apache.org/jira/browse/KAFKA-1811?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Edward Ribeiro reassigned KAFKA-1811: - Assignee: Edward Ribeiro ensuring registered broker host:port is unique -- Key: KAFKA-1811 URL: https://issues.apache.org/jira/browse/KAFKA-1811 Project: Kafka Issue Type: Improvement Reporter: Jun Rao Assignee: Edward Ribeiro Labels: newbie Attachments: KAFKA_1811.patch Currently, we expect each of the registered broker to have a unique host:port pair. However, we don't enforce that, which causes various weird problems. It would be useful to ensure this during broker registration. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 34766: Patch for KAFKA-2229
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94759 --- core/src/main/scala/kafka/server/TopicCommandHelper.scala (line 35) https://reviews.apache.org/r/34766/#comment149361 A lot of this code/functionality exists in kafka.admin.TopicCommand. Is this duplicating a lot of code/functionality? Could they both be refactored to use the same core code? - Grant Henke On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala 252ac813c8df1780c2dc5fa9e698fb43bb6d5cf8 core/src/test/scala/unit/kafka/server/TopicCommandHelperTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/34766/diff/ Testing --- Thanks, Andrii Biletskyi
Re: [DISCUSSION] KIP-29 - Add an IsrPropagateIntervalMs config to KafkaConfig
Not having to deal with configs sounds ideal. I liked the idea of having automatic batching, however that requires broker-to-broker comm support. At some point, I think it will be better to have direct broker - controller communication, rather that using ZK for the purpose. Few more approaches. 1. A broker that initiated IISR change notification will not issue another notification until it receives a metadata upgrade request from controller. At which point it can discard all the ISR change notifications it has queued. 2. ISR change notifications can be limited at each broker by rate rather than time. So, each broker can send up to something like x ISR change notifications in y seconds. On Mon, Aug 10, 2015 at 9:39 AM, Jay Kreps j...@confluent.io wrote: I guess the question, which I think is what Gwen was getting at, is if, rather than making this configurable, it might be possible to make this just work reliably and with the lowest possible latency in some automatic fashion? I raised group commit because that is a way to automatically batch under load. If that doesn't work perhaps there is another way? The challenge as we've seen with obscure configs is that 99% of people can't figure out how to set them so for 99% of people this won't really help them. -Jay On Mon, Aug 10, 2015 at 9:12 AM, Ashish Singh asi...@cloudera.com wrote: Hey Guys, Looks like Jun and Jiangjie have covered the motivation behind the config. To me one would want to set ISR propagation delay to a high number primarily during rolling upgrade. I think the delay one would want to have in ISR propagation is proportional to the cluster size. However, during normal operations a faster ISR propagation is desired. Having a config to expose the delay time provides admin a way to control it, and it will come with a default value so if someone does not want to play with it they can choose not to. @Gwen, does that answer your question? On Sun, Aug 9, 2015 at 3:26 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Jun, Thanks for the detailed explanation. Now I understand. Yes, we might not be able leverage the group commit in this case. When I was testing the patch, I also found a potential use case for the config (not sure if that is a strong use case though). When we rolling upgrade a cluster, if the controller is still running on an old version, the brokers got bounced will create a bunch of ZK path that will not be picked up by the controller until the controller is upgraded to the new version. It might be fine for small clusters, for larger clusters, there could be many ISR change got accumulated before the controller runs on the new version. So if we have the config, when people do rolling upgrade, they can first have the propagation interval to be very large so ISR change will not be propagated. After all the brokers are running on the new version, we can set the propagation interval back to a short value and bounce brokers again. Thanks, Jiangjie (Becket) Qin On Sun, Aug 9, 2015 at 12:22 PM, Jun Rao j...@confluent.io wrote: Jiangjie, Related to group commit, I think what Jay says is the following. Suppose the propagation of ISR is slow and also creates back pressure, you can just propagate the ISR one batch at a time as fast as you can. After one batch is sent, you just collect all ISR changes that have accumulated since the last propagation and propagate them as a batch. This way, the amount of batching happens automatically based on the propagation rate and the delay of propagation is minimized. However, this doesn't quite work if we propagate the ISR changes by writing a notification path in ZK. There is no back pressure: the notification can be written faster than the listener can process it. So the explicit batching that you did essentially is to get around this problem. As for the config, I can see a couple of use cases: (1) We have unit tests that verify the propagation of ISR. To prevent those tests taking too long, we can configure a smaller propagation interval. (2) When there are many partitions and brokers, someone may want to increase the propagation interval to reduce the ZK overhead. I agree that most people likely don't need to change this setting since the default should be good enough. Thanks, Jun On Fri, Aug 7, 2015 at 4:06 PM, Gwen Shapira g...@confluent.io wrote: Maybe Ashish can supply the use-case and tuning advice then :) I'm a -1 on adding new configurations that we can't quite explain. On Fri, Aug 7, 2015 at 3:57 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Hi Gwen, Completely agree with you. I originally just hard coded it to be 10 seconds. Ashish raised this
[jira] [Commented] (KAFKA-2084) byte rate metrics per client ID (producer and consumer)
[ https://issues.apache.org/jira/browse/KAFKA-2084?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680409#comment-14680409 ] Aditya Auradkar commented on KAFKA-2084: [~junrao][~jjkoshy] I think I've addressed all of your comments. Can you guys take another look? byte rate metrics per client ID (producer and consumer) --- Key: KAFKA-2084 URL: https://issues.apache.org/jira/browse/KAFKA-2084 Project: Kafka Issue Type: Sub-task Reporter: Aditya Auradkar Assignee: Aditya Auradkar Labels: quotas Attachments: KAFKA-2084.patch, KAFKA-2084_2015-04-09_18:10:56.patch, KAFKA-2084_2015-04-10_17:24:34.patch, KAFKA-2084_2015-04-21_12:21:18.patch, KAFKA-2084_2015-04-21_12:28:05.patch, KAFKA-2084_2015-05-05_15:27:35.patch, KAFKA-2084_2015-05-05_17:52:02.patch, KAFKA-2084_2015-05-11_16:16:01.patch, KAFKA-2084_2015-05-26_11:50:50.patch, KAFKA-2084_2015-06-02_17:02:00.patch, KAFKA-2084_2015-06-02_17:09:28.patch, KAFKA-2084_2015-06-02_17:10:52.patch, KAFKA-2084_2015-06-04_16:31:22.patch, KAFKA-2084_2015-06-12_10:39:35.patch, KAFKA-2084_2015-06-29_17:53:44.patch, KAFKA-2084_2015-08-04_18:50:51.patch, KAFKA-2084_2015-08-04_19:07:46.patch, KAFKA-2084_2015-08-07_11:27:51.patch We need to be able to track the bytes-in/bytes-out rate on a per-client ID basis. This is necessary for quotas. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-1695) Authenticate connection to Zookeeper
[ https://issues.apache.org/jira/browse/KAFKA-1695?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680417#comment-14680417 ] Parth Brahmbhatt commented on KAFKA-1695: - [~ijuma] Don't have a date, have sent an E-mail to them. Right now I have not included the part that depends on the release so it's not a blocker. The APIs added in the new zkClient release are only required if we want to set the acls on already existing zookeeper nodes. This will be the case for anyone trying to move an existing kafka cluster to secure setup but for fresh installation or users willing to setting the zkAcls on existing kafka nodes manually the current patch should work as is. Authenticate connection to Zookeeper Key: KAFKA-1695 URL: https://issues.apache.org/jira/browse/KAFKA-1695 Project: Kafka Issue Type: Sub-task Components: security Reporter: Jay Kreps Assignee: Parth Brahmbhatt We need to make it possible to secure the Zookeeper cluster Kafka is using. This would make use of the normal authentication ZooKeeper provides. ZooKeeper supports a variety of authentication mechanisms so we will need to figure out what has to be passed in to the zookeeper client. The intention is that when the current round of client work is done it should be possible to run without clients needing access to Zookeeper so all we need here is to make it so that only the Kafka cluster is able to read and write to the Kafka znodes (we shouldn't need to set any kind of acl on a per-znode basis). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2210) KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation.
[ https://issues.apache.org/jira/browse/KAFKA-2210?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680464#comment-14680464 ] Parth Brahmbhatt commented on KAFKA-2210: - [~ijuma] Sorry got busy with some other stuff.Will update the review before EOD. KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation. -- Key: KAFKA-2210 URL: https://issues.apache.org/jira/browse/KAFKA-2210 Project: Kafka Issue Type: Sub-task Components: security Reporter: Parth Brahmbhatt Assignee: Parth Brahmbhatt Fix For: 0.8.3 Attachments: KAFKA-2210.patch, KAFKA-2210_2015-06-03_16:36:11.patch, KAFKA-2210_2015-06-04_16:07:39.patch, KAFKA-2210_2015-07-09_18:00:34.patch, KAFKA-2210_2015-07-14_10:02:19.patch, KAFKA-2210_2015-07-14_14:13:19.patch, KAFKA-2210_2015-07-20_16:42:18.patch, KAFKA-2210_2015-07-21_17:08:21.patch This is the first subtask for Kafka-1688. As Part of this jira we intend to agree on all the public entities, configs and changes to existing kafka classes to allow pluggable authorizer implementation. Please see KIP-11 https://cwiki.apache.org/confluence/display/KAFKA/KIP-11+-+Authorization+Interface for detailed design. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[GitHub] kafka pull request: KAFKA-2134: fix replica offset truncate to beg...
Github user becketqin closed the pull request at: https://github.com/apache/kafka/pull/104 --- 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-2417) Ducktape tests for SSL/TLS
Ismael Juma created KAFKA-2417: -- Summary: Ducktape tests for SSL/TLS Key: KAFKA-2417 URL: https://issues.apache.org/jira/browse/KAFKA-2417 Project: Kafka Issue Type: Sub-task Reporter: Ismael Juma Fix For: 0.8.3 The tests should be complementary to the unit/integration tests written as part of KAFKA-1685. Things to consider: * Upgrade/downgrade to turning on/off SSL * Impact on basic performance * Failure testing * Expired/revoked certificates * Renegotiation Some changes to ducktape may be required for upgrade scenarios. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2143) Replicas get ahead of leader and fail
[
https://issues.apache.org/jira/browse/KAFKA-2143?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680467#comment-14680467
]
ASF GitHub Bot commented on KAFKA-2143:
---
GitHub user becketqin opened a pull request:
https://github.com/apache/kafka/pull/129
KAFKA-2143: fix replica offset truncate to beginning during leader
migration.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/becketqin/kafka KAFKA-2143
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/kafka/pull/129.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 #129
commit 71f8a4716e1f0b4fc2bd88aa30fe38aef8a9f92e
Author: Jiangjie Qin becket@gmail.com
Date: 2015-08-03T02:22:02Z
Fix for KAFKA-2134, fix replica offset truncate to beginning during leader
migration.
Replicas get ahead of leader and fail
-
Key: KAFKA-2143
URL: https://issues.apache.org/jira/browse/KAFKA-2143
Project: Kafka
Issue Type: Bug
Components: replication
Affects Versions: 0.8.2.1
Reporter: Evan Huus
Assignee: Jiangjie Qin
Fix For: 0.8.3
On a cluster of 6 nodes, we recently saw a case where a single
under-replicated partition suddenly appeared, replication lag spiked, and
network IO spiked. The cluster appeared to recover eventually on its own,
Looking at the logs, the thing which failed was partition 7 of the topic
{{background_queue}}. It had an ISR of 1,4,3 and its leader at the time was
3. Here are the interesting log lines:
On node 3 (the leader):
{noformat}
[2015-04-23 16:50:05,879] ERROR [Replica Manager on Broker 3]: Error when
processing fetch request for partition [background_queue,7] offset 3722949957
from follower with correlation id 148185816. Possible cause: Request for
offset 3722949957 but we only have log segments in the range 3648049863 to
3722949955. (kafka.server.ReplicaManager)
[2015-04-23 16:50:05,879] ERROR [Replica Manager on Broker 3]: Error when
processing fetch request for partition [background_queue,7] offset 3722949957
from follower with correlation id 156007054. Possible cause: Request for
offset 3722949957 but we only have log segments in the range 3648049863 to
3722949955. (kafka.server.ReplicaManager)
[2015-04-23 16:50:13,960] INFO Partition [background_queue,7] on broker 3:
Shrinking ISR for partition [background_queue,7] from 1,4,3 to 3
(kafka.cluster.Partition)
{noformat}
Note that both replicas suddenly asked for an offset *ahead* of the available
offsets.
And on nodes 1 and 4 (the replicas) many occurrences of the following:
{noformat}
[2015-04-23 16:50:05,935] INFO Scheduling log segment 3648049863 for log
background_queue-7 for deletion. (kafka.log.Log) (edited)
{noformat}
Based on my reading, this looks like the replicas somehow got *ahead* of the
leader, asked for an invalid offset, got confused, and re-replicated the
entire topic from scratch to recover (this matches our network graphs, which
show 3 sending a bunch of data to 1 and 4).
Taking a stab in the dark at the cause, there appears to be a race condition
where replicas can receive a new offset before the leader has committed it
and is ready to replicate?
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Commented] (KAFKA-1695) Authenticate connection to Zookeeper
[ https://issues.apache.org/jira/browse/KAFKA-1695?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680466#comment-14680466 ] Ismael Juma commented on KAFKA-1695: [~parth.brahmbhatt], thanks. Authenticate connection to Zookeeper Key: KAFKA-1695 URL: https://issues.apache.org/jira/browse/KAFKA-1695 Project: Kafka Issue Type: Sub-task Components: security Reporter: Jay Kreps Assignee: Parth Brahmbhatt We need to make it possible to secure the Zookeeper cluster Kafka is using. This would make use of the normal authentication ZooKeeper provides. ZooKeeper supports a variety of authentication mechanisms so we will need to figure out what has to be passed in to the zookeeper client. The intention is that when the current round of client work is done it should be possible to run without clients needing access to Zookeeper so all we need here is to make it so that only the Kafka cluster is able to read and write to the Kafka znodes (we shouldn't need to set any kind of acl on a per-znode basis). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[GitHub] kafka pull request: KAFKA-2143: fix replica offset truncate to beg...
GitHub user becketqin opened a pull request: https://github.com/apache/kafka/pull/129 KAFKA-2143: fix replica offset truncate to beginning during leader migration. You can merge this pull request into a Git repository by running: $ git pull https://github.com/becketqin/kafka KAFKA-2143 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/kafka/pull/129.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 #129 commit 71f8a4716e1f0b4fc2bd88aa30fe38aef8a9f92e Author: Jiangjie Qin becket@gmail.com Date: 2015-08-03T02:22:02Z Fix for KAFKA-2134, fix replica offset truncate to beginning during leader migration. --- 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-2134) Producer blocked on metric publish
[ https://issues.apache.org/jira/browse/KAFKA-2134?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680531#comment-14680531 ] ASF GitHub Bot commented on KAFKA-2134: --- Github user becketqin closed the pull request at: https://github.com/apache/kafka/pull/104 Producer blocked on metric publish -- Key: KAFKA-2134 URL: https://issues.apache.org/jira/browse/KAFKA-2134 Project: Kafka Issue Type: Bug Components: producer Affects Versions: 0.8.2.1 Environment: debian7, java8 Reporter: Vamsi Subhash Achanta Assignee: Jun Rao Priority: Blocker Hi, We have a REST api to publish to a topic. Yesterday, we started noticing that the producer is not able to produce messages at a good rate and the CLOSE_WAITs of our producer REST app are very high. All the producer REST requests are hence timing out. When we took the thread dump and analysed it, we noticed that the threads are getting blocked on JmxReporter metricChange. Here is the attached stack trace. dw-70 - POST /queues/queue_1/messages #70 prio=5 os_prio=0 tid=0x7f043c8bd000 nid=0x54cf waiting for monitor entry [0x7f04363c7000] java.lang.Thread.State: BLOCKED (on object monitor) at org.apache.kafka.common.metrics.JmxReporter.metricChange(JmxReporter.java:76) - waiting to lock 0x0005c1823860 (a java.lang.Object) at org.apache.kafka.common.metrics.Metrics.registerMetric(Metrics.java:182) - locked 0x0007a5e526c8 (a org.apache.kafka.common.metrics.Metrics) at org.apache.kafka.common.metrics.Sensor.add(Sensor.java:165) - locked 0x0007a5e526e8 (a org.apache.kafka.common.metrics.Sensor) When I looked at the code of metricChange method, it uses a synchronised block on an object resource and it seems that it is held by another. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 34766: Patch for KAFKA-2229
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94756 --- clients/src/main/java/org/apache/kafka/common/protocol/Errors.java (line 81) https://reviews.apache.org/r/34766/#comment149360 Should these new errors be added to kafka.common.ErrorMapping.scala? clients/src/main/java/org/apache/kafka/common/protocol/Errors.java (line 92) https://reviews.apache.org/r/34766/#comment149357 Looks like copy paste error in exception message. core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala (line 20) https://reviews.apache.org/r/34766/#comment149358 Is there a time where we would want to pass through a throwable? (Same for the other Exceptions added) core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala (line 22) https://reviews.apache.org/r/34766/#comment149359 Is this needed? Should we always pass some message? (Same for the other exceptions added? - Grant Henke On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala 252ac813c8df1780c2dc5fa9e698fb43bb6d5cf8 core/src/test/scala/unit/kafka/server/TopicCommandHelperTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/34766/diff/ Testing --- Thanks, Andrii Biletskyi
[GitHub] kafka pull request: KAFKA-1893: Allow regex subscriptions in the n...
GitHub user SinghAsDev opened a pull request: https://github.com/apache/kafka/pull/128 KAFKA-1893: Allow regex subscriptions in the new consumer You can merge this pull request into a Git repository by running: $ git pull https://github.com/SinghAsDev/kafka KAFKA-1893 Alternatively you can review and apply these changes as the patch at: https://github.com/apache/kafka/pull/128.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 #128 commit 7294571b853979f85685ae6a49a4711202a64c04 Author: asingh asi...@cloudera.com Date: 2015-07-28T00:05:48Z KAFKA-1893: Allow regex subscriptions in the new consumer --- 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-1893) Allow regex subscriptions in the new consumer
[
https://issues.apache.org/jira/browse/KAFKA-1893?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680461#comment-14680461
]
ASF GitHub Bot commented on KAFKA-1893:
---
GitHub user SinghAsDev opened a pull request:
https://github.com/apache/kafka/pull/128
KAFKA-1893: Allow regex subscriptions in the new consumer
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/SinghAsDev/kafka KAFKA-1893
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/kafka/pull/128.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 #128
commit 7294571b853979f85685ae6a49a4711202a64c04
Author: asingh asi...@cloudera.com
Date: 2015-07-28T00:05:48Z
KAFKA-1893: Allow regex subscriptions in the new consumer
Allow regex subscriptions in the new consumer
-
Key: KAFKA-1893
URL: https://issues.apache.org/jira/browse/KAFKA-1893
Project: Kafka
Issue Type: Sub-task
Components: consumer
Reporter: Jay Kreps
Assignee: Ashish K Singh
Priority: Critical
Fix For: 0.8.3
The consumer needs to handle subscribing to regular expressions. Presumably
this would be done as a new api,
{code}
void subscribe(java.util.regex.Pattern pattern);
{code}
Some questions/thoughts to work out:
- It should not be possible to mix pattern subscription with partition
subscription.
- Is it allowable to mix this with normal topic subscriptions? Logically
this is okay but a bit complex to implement.
- We need to ensure we regularly update the metadata and recheck our regexes
against the metadata to update subscriptions for new topics that are created
or old topics that are deleted.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
Re: Review Request 34766: Patch for KAFKA-2229
On Aug. 10, 2015, 5:01 p.m., Grant Henke wrote: core/src/main/scala/kafka/server/TopicCommandHelper.scala, line 35 https://reviews.apache.org/r/34766/diff/2/?file=995963#file995963line35 A lot of this code/functionality exists in kafka.admin.TopicCommand. Is this duplicating a lot of code/functionality? Could they both be refactored to use the same core code? IMO no, they cannot be refactored. The main problem is that code in kafka.admin.TopicCommand is very tightly coupled with CLI logic (like printing results, validation inputs), also error handling logic is very different. - Andrii --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/#review94759 --- On June 30, 2015, 1:59 p.m., Andrii Biletskyi wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34766/ --- (Updated June 30, 2015, 1:59 p.m.) Review request for kafka. Bugs: KAFKA-2229 https://issues.apache.org/jira/browse/KAFKA-2229 Repository: kafka Description --- KIP-4 Phase 1 Rebase Diffs - checkstyle/import-control.xml f2e6cec267e67ce8e261341e373718e14a8e8e03 clients/src/main/java/org/apache/kafka/common/ConfigEntry.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/PartitionReplicaAssignment.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/TopicConfigInfo.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b39e9bb53dd51f1ce931691e23aabc9ebaebf1e7 clients/src/main/java/org/apache/kafka/common/protocol/Errors.java 5b898c8f8ad5d0469f469b600c4b2eb13d1fc662 clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java 3dc8b015afd2347a41c9a9dbc02b8e367da5f75f clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java 5d3d52859587ce0981d702f04042b0f6e1bc3704 clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/AlterTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/CreateTopicResponse.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicRequest.java PRE-CREATION clients/src/main/java/org/apache/kafka/common/requests/admin/DeleteTopicResponse.java PRE-CREATION clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java 8b2aca85fa738180e5420985fddc39a4bf9681ea core/src/main/scala/kafka/admin/AdminUtils.scala f06edf41c732a7b794e496d0048b0ce6f897e72b core/src/main/scala/kafka/api/RequestKeys.scala 155cb650e9cffe2c950326cfc25b1480cda819db core/src/main/scala/kafka/common/InvalidPartitionsException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicaAssignmentException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidReplicationFactorException.scala PRE-CREATION core/src/main/scala/kafka/common/InvalidTopicConfigurationException.scala PRE-CREATION core/src/main/scala/kafka/common/ReassignPartitionsInProgressException.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala ad6f05807c61c971e5e60d24bc0c87e989115961 core/src/main/scala/kafka/server/TopicCommandHelper.scala PRE-CREATION core/src/test/scala/unit/kafka/admin/AdminTest.scala 252ac813c8df1780c2dc5fa9e698fb43bb6d5cf8 core/src/test/scala/unit/kafka/server/TopicCommandHelperTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/34766/diff/ Testing --- Thanks, Andrii Biletskyi
Typo on documentation
I have just seen the typo below at http://kafka.apache.org/documentation.html . It's supposed to be JMX instead of JMZ, right? []'s Eddie
[jira] [Updated] (KAFKA-313) Add JSON/CSV output and looping options to ConsumerGroupCommand
[ https://issues.apache.org/jira/browse/KAFKA-313?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ashish K Singh updated KAFKA-313: - Attachment: KAFKA-313_2015-08-10_12:58:38.patch Add JSON/CSV output and looping options to ConsumerGroupCommand --- Key: KAFKA-313 URL: https://issues.apache.org/jira/browse/KAFKA-313 Project: Kafka Issue Type: Improvement Reporter: Dave DeMaagd Assignee: Ashish K Singh Priority: Minor Labels: newbie, patch Fix For: 0.8.3 Attachments: KAFKA-313-2012032200.diff, KAFKA-313.1.patch, KAFKA-313.patch, KAFKA-313_2015-02-23_18:11:32.patch, KAFKA-313_2015-06-24_11:14:24.patch, KAFKA-313_2015-08-05_15:37:32.patch, KAFKA-313_2015-08-05_15:43:00.patch, KAFKA-313_2015-08-10_12:58:38.patch Adds: * '--loop N' - causes the program to loop forever, sleeping for up to N seconds between loops (loop time minus collection time, unless that's less than 0, at which point it will just run again immediately) * '--asjson' - display as a JSON string instead of the more human readable output format. Neither of the above depend on each other (you can loop in the human readable output, or do a single shot execution with JSON output). Existing behavior/output maintained if neither of the above are used. Diff Attached. Impacted files: core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-313) Add JSON/CSV output and looping options to ConsumerGroupCommand
[ https://issues.apache.org/jira/browse/KAFKA-313?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680671#comment-14680671 ] Ashish K Singh commented on KAFKA-313: -- Updated reviewboard https://reviews.apache.org/r/28096/ against branch trunk Add JSON/CSV output and looping options to ConsumerGroupCommand --- Key: KAFKA-313 URL: https://issues.apache.org/jira/browse/KAFKA-313 Project: Kafka Issue Type: Improvement Reporter: Dave DeMaagd Assignee: Ashish K Singh Priority: Minor Labels: newbie, patch Fix For: 0.8.3 Attachments: KAFKA-313-2012032200.diff, KAFKA-313.1.patch, KAFKA-313.patch, KAFKA-313_2015-02-23_18:11:32.patch, KAFKA-313_2015-06-24_11:14:24.patch, KAFKA-313_2015-08-05_15:37:32.patch, KAFKA-313_2015-08-05_15:43:00.patch, KAFKA-313_2015-08-10_12:58:38.patch Adds: * '--loop N' - causes the program to loop forever, sleeping for up to N seconds between loops (loop time minus collection time, unless that's less than 0, at which point it will just run again immediately) * '--asjson' - display as a JSON string instead of the more human readable output format. Neither of the above depend on each other (you can loop in the human readable output, or do a single shot execution with JSON output). Existing behavior/output maintained if neither of the above are used. Diff Attached. Impacted files: core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Typo on documentation
yeppers. JIRA and patch? On Mon, Aug 10, 2015 at 12:36 PM, Edward Ribeiro edward.ribe...@gmail.com wrote: I have just seen the typo below at http://kafka.apache.org/documentation.html . It's supposed to be JMX instead of JMZ, right? []'s Eddie
Re: [DISCUSS] KIP-27 - Conditional Publish
I've been trying to understand what is being proposed in this KIP and I've put down some notes with some feedback from Ben that I wanted to share for feedback. I'm not really following the flow of the thread, since I've read a few sources to get to this, and I apologize for that. Here is how I see it t a high level. There are really two problems being discussed in the context of this KIP: Single writer with failover: Consistent logs Single writer with failover The idea is that at any time there must be at most one publisher active. To get high availability, we can’t rely on a single process to be such a publisher and consequently we need the failover part: if the current active publisher crashes, then another publisher takes over and becomes active. One important issue with scenarios like this is that during transitions from one active publisher to another, there could be races and two publishers end up interleaving messages in a topic/partition/key. Why is this interleaving bad? This is really application specific, but one general way of seeing this is that only one process has the authoritative application state to generate messages to publish. Transitioning from an active publisher to another, typically requires recovering state or performing some kind of coordination. If no such recovery is required, then we are essentially in the multi-writer space. The commit log use case is a general one mentioned in the KIP description. Consistent logs Consistent logs might not be the best term here, but I’m using it to describe the need of having the messages in a topic/partition/key reflecting consistently the state of the application. For example, some applications might be OK with a published sequence: A B B C (e.g., value = 10) in the case the messages are idempotent operations, but others might really require: A B C (e.g., value += 10) if they aren’t idempotent operations. Order and gaps are also an issue, so some applications might be OK with: A C B (e.g., value += x) and skipping B altogether might be ok if B has no side-effects (e.g., operation associated to B has failed). Putting things together The current KIP-27 proposal seems to do a good job with providing a consistent log in the absence of concurrency. It enables publishers to re-publish messages without duplication, which is one requirement for exactly-once semantics. Gaps need to be handled by the publisher. For example, if the publisher publishes A B C (assuming it’s publishing asynchronously and in an open loop), it could have A succeeding but not B and C. In this case, it needs to redo the publish of B and C. It could also have B failing and C succeeding, in which case the publisher repeats B and C. A really nice feature of the current proposal is that it is a simple primitive that enables the implementation of publishers with different delivery guarantees. It doesn’t seem to be well suited to the first problem of implementing a single writer with failover, however. It allows runs in which two producers interleave messages because the mechanism focuses on a single message. The single writer might not even care about duplicates and gaps depending on the application, but it might care that there aren’t two publishers interleaving messages in the Kafka log. A typical way of dealing with these cases is to use a token associated to a lease to fence off the other publishers. For example, to demote an active publisher, another publisher could invoke a demote call and have the ISR leader replace the token. The lease of the token could be done directly with ZooKeeper or via the ISR leader. The condition to publish a message or a batch could be a combination of token verification and offset check. -Flavio On 10 Aug 2015, at 00:15, Jun Rao j...@confluent.io wrote: Couple of other things. A. In the discussion, we talked about the usage of getting the latest high watermark from the broker. Currently, the high watermark in a partition can go back a bit for a short period of time during leader change. So, the high watermark returned in the getOffset api is not 100% accurate. There is a jira (KAFKA-2334) to track this issue. B. The proposal in the wiki is to put the expected offset in every message, even when the messages are compressed. With Jiangjie's proposal of relative offset, the expected offset probably can only be set at the shallow compressed message level. We will need to think this through. Thanks, Jun On Tue, Aug 4, 2015 at 3:05 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Jun, I see. So this only applies to uncompressed messages. Maybe that is fine given most user will probably turn on compression? I think the first approach is a more general approach but from application point of view might harder to implement. I am thinking is it easier for the application simply have one producer for a partition and hash the message to producer. In that case,
Re: Review Request 28096: Patch for KAFKA-313
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/28096/ --- (Updated Aug. 10, 2015, 7:58 p.m.) Review request for kafka, Gwen Shapira, Jarek Cecho, and Joel Koshy. Bugs: KAFKA-313 https://issues.apache.org/jira/browse/KAFKA-313 Repository: kafka Description --- KAFKA-313: Add JSON/CSV output and looping options to ConsumerGroupCommand Diffs (updated) - core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala f23120ede5f9bf0cfaf795c65c9845f42d8784d0 Diff: https://reviews.apache.org/r/28096/diff/ Testing --- Ran ConsumerOffsetChecker with different combinations of --output.format and --loop options. Thanks, Ashish Singh
Re: [DISCUSSION] KIP-29 - Add an IsrPropagateIntervalMs config to KafkaConfig
An easier (although less clean) solution can be a hidden configuration: Avoid adding the new configuration to KafkaConfig's ConfigDef and the docs, but grab a pre-defined parameter anyway (if exists, use the reasonable default if it doesn't). This will allow us to override the value in tests (and emergencies) but avoid confusing users with new user-exposed configurations that they most likely won't need. (The concept is inspired by Oracle's hidden parameters, which serve the same goal but also has support implications) Gwen On Mon, Aug 10, 2015 at 9:39 AM, Jay Kreps j...@confluent.io wrote: I guess the question, which I think is what Gwen was getting at, is if, rather than making this configurable, it might be possible to make this just work reliably and with the lowest possible latency in some automatic fashion? I raised group commit because that is a way to automatically batch under load. If that doesn't work perhaps there is another way? The challenge as we've seen with obscure configs is that 99% of people can't figure out how to set them so for 99% of people this won't really help them. -Jay On Mon, Aug 10, 2015 at 9:12 AM, Ashish Singh asi...@cloudera.com wrote: Hey Guys, Looks like Jun and Jiangjie have covered the motivation behind the config. To me one would want to set ISR propagation delay to a high number primarily during rolling upgrade. I think the delay one would want to have in ISR propagation is proportional to the cluster size. However, during normal operations a faster ISR propagation is desired. Having a config to expose the delay time provides admin a way to control it, and it will come with a default value so if someone does not want to play with it they can choose not to. @Gwen, does that answer your question? On Sun, Aug 9, 2015 at 3:26 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Jun, Thanks for the detailed explanation. Now I understand. Yes, we might not be able leverage the group commit in this case. When I was testing the patch, I also found a potential use case for the config (not sure if that is a strong use case though). When we rolling upgrade a cluster, if the controller is still running on an old version, the brokers got bounced will create a bunch of ZK path that will not be picked up by the controller until the controller is upgraded to the new version. It might be fine for small clusters, for larger clusters, there could be many ISR change got accumulated before the controller runs on the new version. So if we have the config, when people do rolling upgrade, they can first have the propagation interval to be very large so ISR change will not be propagated. After all the brokers are running on the new version, we can set the propagation interval back to a short value and bounce brokers again. Thanks, Jiangjie (Becket) Qin On Sun, Aug 9, 2015 at 12:22 PM, Jun Rao j...@confluent.io wrote: Jiangjie, Related to group commit, I think what Jay says is the following. Suppose the propagation of ISR is slow and also creates back pressure, you can just propagate the ISR one batch at a time as fast as you can. After one batch is sent, you just collect all ISR changes that have accumulated since the last propagation and propagate them as a batch. This way, the amount of batching happens automatically based on the propagation rate and the delay of propagation is minimized. However, this doesn't quite work if we propagate the ISR changes by writing a notification path in ZK. There is no back pressure: the notification can be written faster than the listener can process it. So the explicit batching that you did essentially is to get around this problem. As for the config, I can see a couple of use cases: (1) We have unit tests that verify the propagation of ISR. To prevent those tests taking too long, we can configure a smaller propagation interval. (2) When there are many partitions and brokers, someone may want to increase the propagation interval to reduce the ZK overhead. I agree that most people likely don't need to change this setting since the default should be good enough. Thanks, Jun On Fri, Aug 7, 2015 at 4:06 PM, Gwen Shapira g...@confluent.io wrote: Maybe Ashish can supply the use-case and tuning advice then :) I'm a -1 on adding new configurations that we can't quite explain. On Fri, Aug 7, 2015 at 3:57 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Hi Gwen, Completely agree with you. I originally just hard coded it to be 10 seconds. Ashish raised this requirement in KAFKA-2406 because people might want to ISR changes get propagated quicker. I don't have a good use case myself. Personally I think
Re: [jira] [Commented] (KAFKA-1690) new java producer needs ssl support as a client
Harsha, I am using the code from https://github.com/harshach/kafka/tree/KAFKA-1690-V1 with the latest commit on 25 July which I think corresponds to the latest patch in KAFKA-1690. Is that correct? I have run SSLConsumerTest, SSLProducerSendTest and SSLSelectorTest several times and the only one that fails to complete is SSLSelectorTest.testRenegotiate(). I added some trace to the code and the sequence in the failed run is: 1. Handshake starts 2. Handshake finishes with handshakeStatus=FINISHED appReadBuffer=java.nio.DirectByteBuffer[pos=0 lim=16916 cap=16916] *(there is no data in appReadBuffer)* 3. Several reads and writes occur 4. read() results in handshake() due to renegotiation. appReadBuffer=java.nio.DirectByteBuffer[pos=0 lim=16916 cap=16916] *(no data in appReadBuffer)* 5. Handshake status changes from NEED_TASK to NEED_WRAP to NEED_UNWRAP 6. Unwrap call during handshake results in data in appReadBuffer: handshakeStatus=NEED_UNWRAP appReadBuffer=java.nio.DirectByteBuffer[pos=100 lim=16916 cap=16916] *(data is now in appReadBuffer)* 7. Handshake status changes to NEED_UNWRAP followed by FINISHED. Final call to handshake() returns with handshakeStatus=NOT_HANDSHAKING appReadBuffer=java.nio.DirectByteBuffer[pos=100 lim=16916 cap=16916] *(data is still in appReadBuffer)* Test continues to call selector.poll() forever. But no more data arrives on the channel since all the data in the test has already arrived, and hence SSLTransportLayer.read() is never invoked on the channel and the data that was unwrapped earlier simply remains in appReadBuffer. This is not a scenario that occurs in the other tests where some data always arrives on the network after the handshake() resulting in SSLTransportLayer.read() being invoked. I can recreate this problem quite easily with IBM JDK (it hangs around one in ten runs), so if you require more trace, please let me know. Thank you, Rajini On Mon, Aug 10, 2015 at 5:29 PM, Sriharsha Chintalapani harsh...@fastmail.fm wrote: Thanks for testing out Rajini. I did ran that test in a loop and it never hanged for me. I am hoping you are using the latest patch since the data left over issue is addressed in latest patch. Also if thats an issue SSLConsumerTest and SSLProducerTest will hang too. Did you notice those are having any issues? I am addressing left over reviews will be sending a new patch in a day or two. Thanks, Harsha On August 10, 2015 at 3:35:34 AM, Rajini Sivaram ( rajinisiva...@googlemail.com) wrote: I was running a Kafka cluster with the latest SSL patch over the weekend with IBM JRE, and it has been running fine without any issues. There was light load on the cluster throughout and intermittent heavy load, all using SSL clients. However I am seeing an intermittent unit test hang in org.apache.kafka.common.network.SSLSelectorTest.testRenegotiate(). I am not sure if that is related to the IBM JRE I am using for the build. It looks like data left in appReadBuffer after a handshake may not get processed if no more data arrives on the network, causing the test to loop forever. It will be good if this can be fixed (or at least the test commented out) before the code is committed to avoid breaking the build. Even though there are quite a few outstanding review comments, I do agree that being such a big patch, it will be good to commit the code soon and work on minor issues afterwards. On Mon, Aug 10, 2015 at 12:19 AM, Jun Rao (JIRA) j...@apache.org wrote: [ https://issues.apache.org/jira/browse/KAFKA-1690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14679397#comment-14679397 ] Jun Rao commented on KAFKA-1690: [~rsivaram], this jira is getting pretty close to be committed. Could you test this out on an IBM jvm to see if there is any issue especially with respect to the usage of the sendfile api? new java producer needs ssl support as a client --- Key: KAFKA-1690 URL: https://issues.apache.org/jira/browse/KAFKA-1690 Project: Kafka Issue Type: Sub-task Reporter: Joe Stein Assignee: Sriharsha Chintalapani Fix For: 0.8.3 Attachments: KAFKA-1690.patch, KAFKA-1690.patch, KAFKA-1690_2015-05-10_23:20:30.patch, KAFKA-1690_2015-05-10_23:31:42.patch, KAFKA-1690_2015-05-11_16:09:36.patch, KAFKA-1690_2015-05-12_16:20:08.patch, KAFKA-1690_2015-05-15_07:18:21.patch, KAFKA-1690_2015-05-20_14:54:35.patch, KAFKA-1690_2015-05-21_10:37:08.patch, KAFKA-1690_2015-06-03_18:52:29.patch, KAFKA-1690_2015-06-23_13:18:20.patch, KAFKA-1690_2015-07-20_06:10:42.patch, KAFKA-1690_2015-07-20_11:59:57.patch, KAFKA-1690_2015-07-25_12:10:55.patch -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Typo on documentation
Okay. On Mon, Aug 10, 2015 at 5:21 PM, Gwen Shapira g...@confluent.io wrote: yeppers. JIRA and patch? On Mon, Aug 10, 2015 at 12:36 PM, Edward Ribeiro edward.ribe...@gmail.com wrote: I have just seen the typo below at http://kafka.apache.org/documentation.html . It's supposed to be JMX instead of JMZ, right? []'s Eddie
[jira] [Updated] (KAFKA-2418) Typo on official KAFKA documentation
[ https://issues.apache.org/jira/browse/KAFKA-2418?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Edward Ribeiro updated KAFKA-2418: -- Affects Version/s: 0.8.0 0.8.1 0.8.2.0 Typo on official KAFKA documentation Key: KAFKA-2418 URL: https://issues.apache.org/jira/browse/KAFKA-2418 Project: Kafka Issue Type: Bug Components: website Affects Versions: 0.8.0, 0.8.1, 0.8.2.0 Reporter: Edward Ribeiro Assignee: Edward Ribeiro Priority: Trivial Attachments: KAFKA-2418.patch I have just seen the typo below at http://kafka.apache.org/documentation.html . By the end of the document there's a reference to JMZ instead of JMX. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-2418) Typo on official KAFKA documentation
[ https://issues.apache.org/jira/browse/KAFKA-2418?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Edward Ribeiro updated KAFKA-2418: -- Attachment: KAFKA-2418.patch Adding a patch to web site documentation. Typo on official KAFKA documentation Key: KAFKA-2418 URL: https://issues.apache.org/jira/browse/KAFKA-2418 Project: Kafka Issue Type: Bug Components: website Reporter: Edward Ribeiro Assignee: Edward Ribeiro Priority: Trivial Attachments: KAFKA-2418.patch I have just seen the typo below at http://kafka.apache.org/documentation.html . By the end of the document there's a reference to JMZ instead of JMX. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-2418) Typo on official KAFKA documentation
[ https://issues.apache.org/jira/browse/KAFKA-2418?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Edward Ribeiro updated KAFKA-2418: -- Status: Patch Available (was: Open) Typo on official KAFKA documentation Key: KAFKA-2418 URL: https://issues.apache.org/jira/browse/KAFKA-2418 Project: Kafka Issue Type: Bug Components: website Affects Versions: 0.8.2.0, 0.8.1, 0.8.0 Reporter: Edward Ribeiro Assignee: Edward Ribeiro Priority: Trivial Attachments: KAFKA-2418.patch I have just seen the typo below at http://kafka.apache.org/documentation.html . By the end of the document there's a reference to JMZ instead of JMX. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 33378: Patch for KAFKA-2136
---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/33378/#review94822
---
core/src/main/scala/kafka/server/ReplicaManager.scala (line 312)
https://reviews.apache.org/r/33378/#comment149463
Good point. I think we should commit 2084 before this patch.
While rebasing, I can refactor this patch to pass the throttleTime to the
callback that sends the response. i.e.
quotaManagers.recordAndMaybeThrottle (clientId, value, callback) {
// add to delay queue
// pass in the computed throttle time to the callback.
}
If I do that, I dont need to pass the throttleTime to the responseCallback
in ReplicaManager
- Aditya Auradkar
On July 13, 2015, 8:36 p.m., Aditya Auradkar wrote:
---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/33378/
---
(Updated July 13, 2015, 8:36 p.m.)
Review request for kafka, Joel Koshy and Jun Rao.
Bugs: KAFKA-2136
https://issues.apache.org/jira/browse/KAFKA-2136
Repository: kafka
Description
---
Changes are
- Addressing Joel's comments
- protocol changes to the fetch request and response to return the
throttle_time_ms to clients
- New producer/consumer metrics to expose the avg and max delay time for a
client
- Test cases.
- Addressed Joel's comments
For now the patch will publish a zero delay and return a response
Diffs
-
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
56281ee15cc33dfc96ff64d5b1e596047c7132a4
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
07e65d4a54ba4eef5b787eba3e71cbe9f6a920bd
clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
3dc8b015afd2347a41c9a9dbc02b8e367da5f75f
clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
8686d83aa52e435c6adafbe9ff4bd1602281072a
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
eb8951fba48c335095cc43fc3672de1c733e07ff
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
fabeae3083a8ea55cdacbb9568f3847ccd85bab4
clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java
37ec0b79beafcf5735c386b066eb319fb697eff5
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
419541011d652becf0cda7a5e62ce813cddb1732
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
8b1805d3d2bcb9fe2bacb37d870c3236aa9532c4
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
e3cc1967e407b64cc734548c19e30de700b64ba8
core/src/main/scala/kafka/api/FetchRequest.scala
5b38f8554898e54800abd65a7415dd0ac41fd958
core/src/main/scala/kafka/api/FetchResponse.scala
0b6b33ab6f7a732ff1322b6f48abd4c43e0d7215
core/src/main/scala/kafka/api/ProducerRequest.scala
c866180d3680da03e48d374415f10220f6ca68c4
core/src/main/scala/kafka/api/ProducerResponse.scala
5d1fac4cb8943f5bfaa487f8e9d9d2856efbd330
core/src/main/scala/kafka/consumer/SimpleConsumer.scala
c16f7edd322709060e54c77eb505c44cbd77a4ec
core/src/main/scala/kafka/server/AbstractFetcherThread.scala
83fc47417dd7fe4edf030217fa7fd69d99b170b0
core/src/main/scala/kafka/server/DelayedFetch.scala
de6cf5bdaa0e70394162febc63b50b55ca0a92db
core/src/main/scala/kafka/server/DelayedProduce.scala
05078b24ef28f2f4e099afa943e43f1d00359fda
core/src/main/scala/kafka/server/KafkaApis.scala
d63bc18d795a6f0e6994538ca55a9a46f7fb8ffd
core/src/main/scala/kafka/server/OffsetManager.scala
5cca85cf727975f6d3acb2223fd186753ad761dc
core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
b31b432a226ba79546dd22ef1d2acbb439c2e9a3
core/src/main/scala/kafka/server/ReplicaManager.scala
59c9bc3ac3a8afc07a6f8c88c5871304db588d17
core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala
5717165f2344823fabe8f7cfafae4bb8af2d949a
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
00d59337a99ac135e8689bd1ecd928f7b1423d79
Diff: https://reviews.apache.org/r/33378/diff/
Testing
---
New tests added
Thanks,
Aditya Auradkar
Re: Review Request 36858: Patch for KAFKA-2120
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/36858/#review94824 --- clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java (line 135) https://reviews.apache.org/r/36858/#comment149464 When we code a method that may or may not execute a given operation, it's usually a good practice to name it ``maybeExpire``. There's a couple of examples of maybe* in Kafka code base already. wdyt? - Edward Ribeiro On July 29, 2015, 10:58 p.m., Mayuresh Gharat wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/36858/ --- (Updated July 29, 2015, 10:58 p.m.) Review request for kafka. Bugs: KAFKA-2120 https://issues.apache.org/jira/browse/KAFKA-2120 Repository: kafka Description --- Patch for Kip-19 : Added RequestTimeOut and MaxBlockTimeOut Solved compile error Addressed Jason's comments for Kip-19 Diffs - clients/src/main/java/org/apache/kafka/clients/ClientRequest.java dc8f0f115bcda893c95d17c0a57be8d14518d034 clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java 2c421f42ed3fc5d61cf9c87a7eaa7bb23e26f63b clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java 15d00d4e484bb5d51a9ae6857ed6e024a2cc1820 clients/src/main/java/org/apache/kafka/clients/KafkaClient.java 7ab2503794ff3aab39df881bd9fbae6547827d3b clients/src/main/java/org/apache/kafka/clients/NetworkClient.java 0e51d7bd461d253f4396a5b6ca7cd391658807fa clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java 70377ae2fa46deb381139d28590ce6d4115e1adc clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java 923ff999d1b04718ddd9a9132668446525bf62f3 clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java 9517d9d0cd480d5ba1d12f1fde7963e60528d2f8 clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java 03b8dd23df63a8d8a117f02eabcce4a2d48c44f7 clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java aa264202f2724907924985a5ecbe74afc4c6c04b clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java 4cb1e50d6c4ed55241aeaef1d3af09def5274103 clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java a152bd7697dca55609a9ec4cfe0a82c10595fbc3 clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java 06182db1c3a5da85648199b4c0c98b80ea7c6c0c clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java 0baf16e55046a2f49f6431e01d52c323c95eddf0 clients/src/main/java/org/apache/kafka/common/network/Selector.java aaf60c98c2c0f4513a8d65ee0db67953a529d598 clients/src/test/java/org/apache/kafka/clients/MockClient.java d9c97e966c0e2fb605b67285f4275abb89f8813e clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 43238ceaad0322e39802b615bb805b895336a009 clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java 2c693824fa53db1e38766b8c66a0ef42ef9d0f3a clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java 5b2e4ffaeab7127648db608c179703b27b577414 clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java 8b1805d3d2bcb9fe2bacb37d870c3236aa9532c4 Diff: https://reviews.apache.org/r/36858/diff/ Testing --- Thanks, Mayuresh Gharat
Re: Review Request 36858: Patch for KAFKA-2120
---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/36858/#review94817
---
clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java (line 140)
https://reviews.apache.org/r/36858/#comment149453
Just out of curiosity: why this relience on use ``Iterable`` instead of
using ``Collection`` or even ``List``. Didn't get why to use the highest
interface is preferrable here. Again, just out of curiosity.
clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java (line 141)
https://reviews.apache.org/r/36858/#comment149450
As jdk 1.7 is the current version then this line can be simplified to use
diamond operators:
ListString nodeIds = new LinkedList();
clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java (line 142)
https://reviews.apache.org/r/36858/#comment149449
this if condition is unnecessary: if the map is empty the for-loop will NOT
iterate a single time.
clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java (line 154)
https://reviews.apache.org/r/36858/#comment149452
nit: I would wrap this on Collections.unmodifiableList() to make this List
virtually immutable, in this particular case.
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
(line 209)
https://reviews.apache.org/r/36858/#comment149454
Use diamond operators:
``ListRecordBatch expiredBatches = new ArrayList();``
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
(line 216)
https://reviews.apache.org/r/36858/#comment149460
The lines can be simplified (i.e., more readable) a tidy bit. You can
either write:
```
for (IteratorRecordBatch it = dq.iterator; it.hasNext; ) {
RecordBatch batch = batchIterator.next();
Node leader = cluster.leaderFor(topicAndPartition);
if (batch != null leader == null) {
// check if the batch is expired
if (batch.expire(requestTimeout, now)) {
expiredBatches.add(batch);
count++;
it.remove();
deallocate(batch);
}
}
}
```
Or even simplier:
```
for (RecordBatch batch : dq) {
Node leader = cluster.leaderFor(topicAndPartition);
if (batch != null leader == null) {
// check if the batch is expired
if (batch.expire(requestTimeout, now)) {
expiredBatches.add(batch);
count++;
dq.remove(batch);
deallocate(batch);
}
}
}
```
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
(line 236)
https://reviews.apache.org/r/36858/#comment149461
Again, we could wrap it in ``Collections.unmodifiableList``? (ps: I am
asking, not stating).
- Edward Ribeiro
On July 29, 2015, 10:58 p.m., Mayuresh Gharat wrote:
---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/36858/
---
(Updated July 29, 2015, 10:58 p.m.)
Review request for kafka.
Bugs: KAFKA-2120
https://issues.apache.org/jira/browse/KAFKA-2120
Repository: kafka
Description
---
Patch for Kip-19 : Added RequestTimeOut and MaxBlockTimeOut
Solved compile error
Addressed Jason's comments for Kip-19
Diffs
-
clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
dc8f0f115bcda893c95d17c0a57be8d14518d034
clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
2c421f42ed3fc5d61cf9c87a7eaa7bb23e26f63b
clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java
15d00d4e484bb5d51a9ae6857ed6e024a2cc1820
clients/src/main/java/org/apache/kafka/clients/KafkaClient.java
7ab2503794ff3aab39df881bd9fbae6547827d3b
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
0e51d7bd461d253f4396a5b6ca7cd391658807fa
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
70377ae2fa46deb381139d28590ce6d4115e1adc
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
923ff999d1b04718ddd9a9132668446525bf62f3
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
9517d9d0cd480d5ba1d12f1fde7963e60528d2f8
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
03b8dd23df63a8d8a117f02eabcce4a2d48c44f7
clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java
aa264202f2724907924985a5ecbe74afc4c6c04b
[jira] [Updated] (KAFKA-2281) org.apache.kafka.clients.producer.internals.ErrorLoggingCallback holds unnecessary byte[] value
[ https://issues.apache.org/jira/browse/KAFKA-2281?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ewen Cheslack-Postava updated KAFKA-2281: - Fix Version/s: 0.8.3 org.apache.kafka.clients.producer.internals.ErrorLoggingCallback holds unnecessary byte[] value --- Key: KAFKA-2281 URL: https://issues.apache.org/jira/browse/KAFKA-2281 Project: Kafka Issue Type: Bug Components: producer Affects Versions: 0.8.2.1 Reporter: TAO XIAO Assignee: TAO XIAO Fix For: 0.8.3 Attachments: KAFKA-2281_2015-06-25.patch org.apache.kafka.clients.producer.internals.ErrorLoggingCallback is constructed with byte[] value as one of the input. It holds the reference to the value until it finishes its lifecycle. The value is not used except for logging its size. This behavior causes unnecessary memory consumption. The fix is to keep reference to the value size instead of value itself when logAsString is false -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 28096: Patch for KAFKA-313
On July 29, 2015, 6:35 p.m., Gwen Shapira wrote: core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala, lines 237-242 https://reviews.apache.org/r/28096/diff/4/?file=991387#file991387line237 These look identical - copy/paste error? Ashish Singh wrote: Not really. There is some difference, None has %s, %s format, while CSV has %s,%s format. The KIP (https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=56852556) doesn't mention NONE. Since CSV and NONE are so similar (just a matter of an extra space), does it make sense to just drop NONE? (which was my expectation, given the KIP) - Gwen --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/28096/#review93489 --- On Aug. 10, 2015, 7:58 p.m., Ashish Singh wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/28096/ --- (Updated Aug. 10, 2015, 7:58 p.m.) Review request for kafka, Gwen Shapira, Jarek Cecho, and Joel Koshy. Bugs: KAFKA-313 https://issues.apache.org/jira/browse/KAFKA-313 Repository: kafka Description --- KAFKA-313: Add JSON/CSV output and looping options to ConsumerGroupCommand Diffs - core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala f23120ede5f9bf0cfaf795c65c9845f42d8784d0 Diff: https://reviews.apache.org/r/28096/diff/ Testing --- Ran ConsumerOffsetChecker with different combinations of --output.format and --loop options. Thanks, Ashish Singh
Re: [DISCUSSION] KIP-29 - Add an IsrPropagateIntervalMs config to KafkaConfig
Jay, Agreed that we should avoid such configurations if possible. Besides that I think we should also try to avoid super complicated implementation that can potentially cause even more problems. Ashish, For (1), it is basically the group commit Jay recommended. The implementation might be a little bit tricky, though. What if broker 2 received a UpdateMetadataRequest which was triggered by broker 1? For (2), the worst case propagation delay is still y seconds. Is it essentially different from hard code the propagation interval to y seconds? The goal we want to achieve here are: 1. If there are small number of ISR changes, we want them to propagate immediately. 2. If there are large number of ISR changes, we want to batch and throttle them. Maybe we can do the following: 1. At time T0, if an ISR change occurs, we plan to send the ISR change at T + 100ms. 2. If no more ISR change occurs between T0 and T0+100ms, we send the ISR notification. If a new ISR change occur at T1 (T0 T1 T0+100ms), we plan to send at T1+100ms. So every newly added ISR change delay the send a little bit. 3. If current time is at T0 + 1000ms, we always send the ISR notification. In most cases, an ISR change will be propagated within 100ms during small number of ISR change. and during peak we batch every second. There might be small chance that an ISR change occur every 100 ms and we end up sending 10 notification / second. But that perhaps isn't too bad. Thoughts? Jiangjie (Becket) Qin On Mon, Aug 10, 2015 at 10:15 AM, Ashish Singh asi...@cloudera.com wrote: Not having to deal with configs sounds ideal. I liked the idea of having automatic batching, however that requires broker-to-broker comm support. At some point, I think it will be better to have direct broker - controller communication, rather that using ZK for the purpose. Few more approaches. 1. A broker that initiated IISR change notification will not issue another notification until it receives a metadata upgrade request from controller. At which point it can discard all the ISR change notifications it has queued. 2. ISR change notifications can be limited at each broker by rate rather than time. So, each broker can send up to something like x ISR change notifications in y seconds. On Mon, Aug 10, 2015 at 9:39 AM, Jay Kreps j...@confluent.io wrote: I guess the question, which I think is what Gwen was getting at, is if, rather than making this configurable, it might be possible to make this just work reliably and with the lowest possible latency in some automatic fashion? I raised group commit because that is a way to automatically batch under load. If that doesn't work perhaps there is another way? The challenge as we've seen with obscure configs is that 99% of people can't figure out how to set them so for 99% of people this won't really help them. -Jay On Mon, Aug 10, 2015 at 9:12 AM, Ashish Singh asi...@cloudera.com wrote: Hey Guys, Looks like Jun and Jiangjie have covered the motivation behind the config. To me one would want to set ISR propagation delay to a high number primarily during rolling upgrade. I think the delay one would want to have in ISR propagation is proportional to the cluster size. However, during normal operations a faster ISR propagation is desired. Having a config to expose the delay time provides admin a way to control it, and it will come with a default value so if someone does not want to play with it they can choose not to. @Gwen, does that answer your question? On Sun, Aug 9, 2015 at 3:26 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: Jun, Thanks for the detailed explanation. Now I understand. Yes, we might not be able leverage the group commit in this case. When I was testing the patch, I also found a potential use case for the config (not sure if that is a strong use case though). When we rolling upgrade a cluster, if the controller is still running on an old version, the brokers got bounced will create a bunch of ZK path that will not be picked up by the controller until the controller is upgraded to the new version. It might be fine for small clusters, for larger clusters, there could be many ISR change got accumulated before the controller runs on the new version. So if we have the config, when people do rolling upgrade, they can first have the propagation interval to be very large so ISR change will not be propagated. After all the brokers are running on the new version, we can set the propagation interval back to a short value and bounce brokers again. Thanks, Jiangjie (Becket) Qin On Sun, Aug 9, 2015 at 12:22 PM, Jun Rao j...@confluent.io wrote: Jiangjie, Related to group commit, I think what Jay says is the following. Suppose the propagation of ISR is slow and also creates
[jira] [Updated] (KAFKA-2084) byte rate metrics per client ID (producer and consumer)
[ https://issues.apache.org/jira/browse/KAFKA-2084?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aditya A Auradkar updated KAFKA-2084: - Attachment: KAFKA-2084_2015-08-10_13:48:50.patch byte rate metrics per client ID (producer and consumer) --- Key: KAFKA-2084 URL: https://issues.apache.org/jira/browse/KAFKA-2084 Project: Kafka Issue Type: Sub-task Reporter: Aditya Auradkar Assignee: Aditya Auradkar Labels: quotas Attachments: KAFKA-2084.patch, KAFKA-2084_2015-04-09_18:10:56.patch, KAFKA-2084_2015-04-10_17:24:34.patch, KAFKA-2084_2015-04-21_12:21:18.patch, KAFKA-2084_2015-04-21_12:28:05.patch, KAFKA-2084_2015-05-05_15:27:35.patch, KAFKA-2084_2015-05-05_17:52:02.patch, KAFKA-2084_2015-05-11_16:16:01.patch, KAFKA-2084_2015-05-26_11:50:50.patch, KAFKA-2084_2015-06-02_17:02:00.patch, KAFKA-2084_2015-06-02_17:09:28.patch, KAFKA-2084_2015-06-02_17:10:52.patch, KAFKA-2084_2015-06-04_16:31:22.patch, KAFKA-2084_2015-06-12_10:39:35.patch, KAFKA-2084_2015-06-29_17:53:44.patch, KAFKA-2084_2015-08-04_18:50:51.patch, KAFKA-2084_2015-08-04_19:07:46.patch, KAFKA-2084_2015-08-07_11:27:51.patch, KAFKA-2084_2015-08-10_13:48:50.patch We need to be able to track the bytes-in/bytes-out rate on a per-client ID basis. This is necessary for quotas. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 33049: Patch for KAFKA-2084
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 10, 2015, 8:49 p.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description (updated) --- Updated patch for quotas. This patch does the following: 1. Add per-client metrics for both producer and consumers 2. Add configuration for quotas 3. Compute delay times in the metrics package and return the delay times in QuotaViolationException 4. Add a DelayQueue in KafkaApi's that can be used to throttle any type of request. Implemented request throttling for produce and fetch requests. 5. Added unit and integration test cases for both producer and consumer 6. This doesn't include a system test. There is a separate ticket for that 7. Fixed KAFKA-2191 - (Included fix from : https://reviews.apache.org/r/34418/ ) Addressed comments from Joel and Jun Diffs - clients/src/main/java/org/apache/kafka/common/metrics/Quota.java d82bb0c055e631425bc1ebbc7d387baac76aeeaa clients/src/main/java/org/apache/kafka/common/metrics/QuotaViolationException.java a451e5385c9eca76b38b425e8ac856b2715fcffe clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java ca823fd4639523018311b814fde69b6177e73b97 clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java 98429da34418f7f1deba1b5e44e2e6025212edb3 clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java 544e120594de78c43581a980b1e4087b4fb98ccb clients/src/test/java/org/apache/kafka/common/utils/MockTime.java core/src/main/scala/kafka/server/ClientQuotaManager.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala 7ea509c2c41acc00430c74e025e069a833aac4e7 core/src/main/scala/kafka/server/KafkaConfig.scala dbe170f87331f43e2dc30165080d2cb7dfe5fdbf core/src/main/scala/kafka/server/KafkaServer.scala 84d4730ac634f9a5bf12a656e422fea03ad72da8 core/src/main/scala/kafka/server/ReplicaManager.scala 795220e7f63d163be90738b4c1a39687b44c1395 core/src/main/scala/kafka/server/ThrottledResponse.scala PRE-CREATION core/src/main/scala/kafka/utils/ShutdownableThread.scala fc226c863095b7761290292cd8755cd7ad0f155c core/src/test/scala/integration/kafka/api/QuotasTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala f32d206d3f52f3f9f4d649c213edd7058f4b6150 core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/33049/diff/ Testing --- Thanks, Aditya Auradkar
[jira] [Updated] (KAFKA-1929) Convert core kafka module to use the errors in org.apache.kafka.common.errors
[ https://issues.apache.org/jira/browse/KAFKA-1929?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jeff Holoman updated KAFKA-1929: Assignee: Grant Henke (was: Jeff Holoman) Convert core kafka module to use the errors in org.apache.kafka.common.errors - Key: KAFKA-1929 URL: https://issues.apache.org/jira/browse/KAFKA-1929 Project: Kafka Issue Type: Improvement Reporter: Jay Kreps Assignee: Grant Henke Attachments: KAFKA-1929.patch With the introduction of the common package there are now a lot of errors duplicated in both the common package and in the server. We should refactor the server code (but not the scala clients) to switch over to the exceptions in common. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 35437: Patch for KAFKA-2202
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/35437/#review94866 --- Ship it! Ship It! - Guozhang Wang On June 14, 2015, 11:27 a.m., Manikumar Reddy O wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/35437/ --- (Updated June 14, 2015, 11:27 a.m.) Review request for kafka. Bugs: KAFKA-2202 https://issues.apache.org/jira/browse/KAFKA-2202 Repository: kafka Description --- while computing stats, consumerTimeoutMs is considered only during ConsumerTimeout exception senarios; This should solve KAFKA-1828 also Diffs - core/src/main/scala/kafka/tools/ConsumerPerformance.scala 903318d15893af08104a97499798c9ad0ba98013 Diff: https://reviews.apache.org/r/35437/diff/ Testing --- Thanks, Manikumar Reddy O
[jira] [Updated] (KAFKA-2202) ConsumerPerformance reports a throughput much higher than the actual one
[
https://issues.apache.org/jira/browse/KAFKA-2202?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Guozhang Wang updated KAFKA-2202:
-
Resolution: Fixed
Fix Version/s: 0.8.3
Status: Resolved (was: Patch Available)
ConsumerPerformance reports a throughput much higher than the actual one
Key: KAFKA-2202
URL: https://issues.apache.org/jira/browse/KAFKA-2202
Project: Kafka
Issue Type: Bug
Components: tools
Affects Versions: 0.8.2.0
Reporter: Micael Capitão
Assignee: Manikumar Reddy
Priority: Minor
Fix For: 0.8.3
Attachments: KAFKA-2202.patch
I've been using the kafka.tools.ConsumerPerformance tool for some
benchmarking until in one of my tests I got a throughput much higher than the
supported by my network interface.
The test consisted in consuming around ~4900 MB from one topic using one
consumer with one thread. The reported throughput reported was ~1400 MB/s
which surpasses the 10 Gbps of the network. The time for the whole operation
was ~8 seconds, which should correspond to a throughput of ~612 MB/s.
Digging the ConsumerPerformance code, I've found this at line 73:
{code:java}
val elapsedSecs = (endMs - startMs - config.consumerConfig.consumerTimeoutMs)
/ 1000.0
{code}
The {{consumerTimeoutMs}} defined as 5000 at line 131 is always considered
leading to wrong results.
This bug seems to be related to this one
[https://issues.apache.org/jira/browse/KAFKA-1828]
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Resolved] (KAFKA-1828) [ConsumerPerformance] the test result is negative number
[ https://issues.apache.org/jira/browse/KAFKA-1828?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Guozhang Wang resolved KAFKA-1828. -- Resolution: Duplicate [ConsumerPerformance] the test result is negative number Key: KAFKA-1828 URL: https://issues.apache.org/jira/browse/KAFKA-1828 Project: Kafka Issue Type: Bug Components: tools Reporter: maji2014 Priority: Minor the test result like: 2014-12-23 19:15:15:329, 2014-12-23 19:15:15:400, 1048576, 0.0790, -0.1842, 1000, -2331.0023 the reason why the result is negative number is that the running time is less than consumer.timeout.ms, but the user doesn't know the reason, so add judgement for that. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2202) ConsumerPerformance reports a throughput much higher than the actual one
[
https://issues.apache.org/jira/browse/KAFKA-2202?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14681204#comment-14681204
]
Guozhang Wang commented on KAFKA-2202:
--
Thanks for the patch [~omkreddy], +1 and committed to trunk.
ConsumerPerformance reports a throughput much higher than the actual one
Key: KAFKA-2202
URL: https://issues.apache.org/jira/browse/KAFKA-2202
Project: Kafka
Issue Type: Bug
Components: tools
Affects Versions: 0.8.2.0
Reporter: Micael Capitão
Assignee: Manikumar Reddy
Priority: Minor
Fix For: 0.8.3
Attachments: KAFKA-2202.patch
I've been using the kafka.tools.ConsumerPerformance tool for some
benchmarking until in one of my tests I got a throughput much higher than the
supported by my network interface.
The test consisted in consuming around ~4900 MB from one topic using one
consumer with one thread. The reported throughput reported was ~1400 MB/s
which surpasses the 10 Gbps of the network. The time for the whole operation
was ~8 seconds, which should correspond to a throughput of ~612 MB/s.
Digging the ConsumerPerformance code, I've found this at line 73:
{code:java}
val elapsedSecs = (endMs - startMs - config.consumerConfig.consumerTimeoutMs)
/ 1000.0
{code}
The {{consumerTimeoutMs}} defined as 5000 at line 131 is always considered
leading to wrong results.
This bug seems to be related to this one
[https://issues.apache.org/jira/browse/KAFKA-1828]
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
[jira] [Created] (KAFKA-2419) Allow certain Sensors to be garbage collected after inactivity
Aditya Auradkar created KAFKA-2419: -- Summary: Allow certain Sensors to be garbage collected after inactivity Key: KAFKA-2419 URL: https://issues.apache.org/jira/browse/KAFKA-2419 Project: Kafka Issue Type: New Feature Reporter: Aditya Auradkar Assignee: Aditya Auradkar Implement a feature to remove certain sensors after a certain period of inactivity (perhaps configurable). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 33049: Patch for KAFKA-2084
On Aug. 6, 2015, 4:17 p.m., Jun Rao wrote: A few more comments. We need to be careful with sensors at the client-id level. Clients can come and go (e.g. console consumer). We probably don't want to hold sensors that are not longer actively used since it takes memory. So, we will need some way of removing inactive sensors. Not sure if we should add this at the metric level or at the quota level. Jun Rao wrote: Did you address the comment on removing inactive sensors? Ah, I missed this comment. Good point.. we should be removing these sensor objects. I think we should handle this in the Metrics library itself.. it would be nice to support sensors that can be garbage collected after a certain period of inactivity (if the sensor is marked as eligible for removal). The new metrics library does not support removal of sensors right now so I filed a ticket as followup since it might need a bit more discussion: https://issues.apache.org/jira/browse/KAFKA-2419 - Aditya --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/#review94412 --- On Aug. 10, 2015, 8:49 p.m., Aditya Auradkar wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 10, 2015, 8:49 p.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description --- Updated patch for quotas. This patch does the following: 1. Add per-client metrics for both producer and consumers 2. Add configuration for quotas 3. Compute delay times in the metrics package and return the delay times in QuotaViolationException 4. Add a DelayQueue in KafkaApi's that can be used to throttle any type of request. Implemented request throttling for produce and fetch requests. 5. Added unit and integration test cases for both producer and consumer 6. This doesn't include a system test. There is a separate ticket for that 7. Fixed KAFKA-2191 - (Included fix from : https://reviews.apache.org/r/34418/ ) Addressed comments from Joel and Jun Diffs - clients/src/main/java/org/apache/kafka/common/metrics/Quota.java d82bb0c055e631425bc1ebbc7d387baac76aeeaa clients/src/main/java/org/apache/kafka/common/metrics/QuotaViolationException.java a451e5385c9eca76b38b425e8ac856b2715fcffe clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java ca823fd4639523018311b814fde69b6177e73b97 clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java 98429da34418f7f1deba1b5e44e2e6025212edb3 clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java 544e120594de78c43581a980b1e4087b4fb98ccb clients/src/test/java/org/apache/kafka/common/utils/MockTime.java core/src/main/scala/kafka/server/ClientQuotaManager.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala 7ea509c2c41acc00430c74e025e069a833aac4e7 core/src/main/scala/kafka/server/KafkaConfig.scala dbe170f87331f43e2dc30165080d2cb7dfe5fdbf core/src/main/scala/kafka/server/KafkaServer.scala 84d4730ac634f9a5bf12a656e422fea03ad72da8 core/src/main/scala/kafka/server/ReplicaManager.scala 795220e7f63d163be90738b4c1a39687b44c1395 core/src/main/scala/kafka/server/ThrottledResponse.scala PRE-CREATION core/src/main/scala/kafka/utils/ShutdownableThread.scala fc226c863095b7761290292cd8755cd7ad0f155c core/src/test/scala/integration/kafka/api/QuotasTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala f32d206d3f52f3f9f4d649c213edd7058f4b6150 core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/33049/diff/ Testing --- Thanks, Aditya Auradkar
[jira] [Updated] (KAFKA-2419) Allow certain Sensors to be garbage collected after inactivity
[ https://issues.apache.org/jira/browse/KAFKA-2419?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Aditya Auradkar updated KAFKA-2419: --- Description: Currently, metrics cannot be removed once registered. Implement a feature to remove certain sensors after a certain period of inactivity (perhaps configurable). was:Implement a feature to remove certain sensors after a certain period of inactivity (perhaps configurable). Allow certain Sensors to be garbage collected after inactivity -- Key: KAFKA-2419 URL: https://issues.apache.org/jira/browse/KAFKA-2419 Project: Kafka Issue Type: New Feature Reporter: Aditya Auradkar Assignee: Aditya Auradkar Labels: quotas Currently, metrics cannot be removed once registered. Implement a feature to remove certain sensors after a certain period of inactivity (perhaps configurable). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 33049: Patch for KAFKA-2084
On Aug. 6, 2015, 2:02 a.m., Jun Rao wrote: clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java, lines 135-139 https://reviews.apache.org/r/33049/diff/18/?file=1032143#file1032143line135 Is that calculation here right? Based on the calculation in Throttler, it should be sth like the following. metricValue/quota.bound() - windowSize Also, the window size is calculated as config.sampels() * config.timeWindows(), which is inaccurate. The last window is not complete. So we need to take the current time into consideration. Finally, it seems that delayTime only makes sense for rates and not for anything else. Perhaps we can at least add a comment. Aditya Auradkar wrote: Hey Jun - Can you elaborate a little? How would we use the current time exactly? It is not clear to me how subtracting the windowSize (time unit) from a fraction (metricValue/quota.bound()) gives the right delay. I also added a comment for delayTime making sense for rates only. Aditya Auradkar wrote: I dug into the Throttler code a bit. Basically - The metricValue is the absolute actual value and not a rate. Throttler delay: (currentValueOfCounter/quotaRate) - elapsedTime Current Sensor delay: ((currentRateValue - quotaRate)/quotaRate) * elapsedTime Lets take an example (all in seconds): quotaRate = 10QPS elapedSec = 20 (Lets say 21 windows of 1 second each. The last second has not started yet) currentValueOfCounter = 250 currentRate = (250/20) = 12.5 (assuming 20 elapsed seconds. Current second may be incomplete) Throttler Formula delay = (currentValueOfCounter/quotaRate) - elapsedTime = (250/10) - 20 = 5 second delay Current Sensor delay = ((currentRateValue - quotaRate)/quotaRate) * numWindows * windowSize = ((12.5 - 10)/10) * 21 windows * 1 second window = 2.5/10 * 21 = 21/4 = 5.25 second delay I think the only discrepancy is in the elapsedTime. The last window is not complete but should be very similar because we configure many small samples. The rate calculation is done inside Rate.java which does not expose the exact elapsed time and the actual counter value. Let's examine how the rate changes because of this: The currentRate returned by the sample will still be 12.5. However, Sensor.java uses 21 as the value because we have 21 windows configured. If we got the exact elapsed time, we would use 20 elapsed seconds Potential Sensor delay = ((currentRateValue - quotaRate)/quotaRate) * elapsedTime = ((12.5 - 10)/10) * 20 = 2.5/10 * 20 = 20/4 = 5 second delay The values are now exactly identical. Given that 5.25 second delay is slightly more pessimistic, we basically throttle a bit more aggressively if the last window is not complete. We can improve this in 2 ways: - Expose elapsedTime and raw metric value from Stat.java - A better solution (IMO), is to throw QuotaViolationException from Rate.java itself. We can also add the delayTime within the Rate class since it does not make sense for other metrics. Finally, We should consider refactoring Throttler.scala to use the Rate metric for its quota computation since the formulae will be identical and only performed in 1 place. I can tackle all of this pretty quickly in a followup patch if that is acceptable. Jun Rao wrote: Got it. I think your calculation is correct. Perhaps we can add a comment on how the formula is derived. However, it does seem that rate includes the last partial window in calculating the rate (see Rate.measure()). So, in your formula, elapsedTime should probably be convert(now - stat.oldest(now).lastWindowMs) where now is passed in from record(). Thanks Jun. I'll add a comment. You are right about the elapsedTime using the last partial window. I used 20 seconds in my example for simplicity. - Aditya --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/#review94333 --- On Aug. 10, 2015, 8:49 p.m., Aditya Auradkar wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 10, 2015, 8:49 p.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description --- Updated patch for quotas. This patch does the following: 1. Add per-client metrics for both producer and consumers 2. Add configuration for quotas 3. Compute delay times in the metrics package and return the delay times in QuotaViolationException 4. Add
[jira] [Created] (KAFKA-2420) Merge the Throttle time computation for Quotas and Throttler
Aditya Auradkar created KAFKA-2420: -- Summary: Merge the Throttle time computation for Quotas and Throttler Key: KAFKA-2420 URL: https://issues.apache.org/jira/browse/KAFKA-2420 Project: Kafka Issue Type: Improvement Reporter: Aditya Auradkar Assignee: Aditya Auradkar Our quota implementation computes Throttle time separately from Throttler.scala. Unify the calculation -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2084) byte rate metrics per client ID (producer and consumer)
[ https://issues.apache.org/jira/browse/KAFKA-2084?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680741#comment-14680741 ] Aditya A Auradkar commented on KAFKA-2084: -- Updated reviewboard https://reviews.apache.org/r/33049/diff/ against branch origin/trunk byte rate metrics per client ID (producer and consumer) --- Key: KAFKA-2084 URL: https://issues.apache.org/jira/browse/KAFKA-2084 Project: Kafka Issue Type: Sub-task Reporter: Aditya Auradkar Assignee: Aditya Auradkar Labels: quotas Attachments: KAFKA-2084.patch, KAFKA-2084_2015-04-09_18:10:56.patch, KAFKA-2084_2015-04-10_17:24:34.patch, KAFKA-2084_2015-04-21_12:21:18.patch, KAFKA-2084_2015-04-21_12:28:05.patch, KAFKA-2084_2015-05-05_15:27:35.patch, KAFKA-2084_2015-05-05_17:52:02.patch, KAFKA-2084_2015-05-11_16:16:01.patch, KAFKA-2084_2015-05-26_11:50:50.patch, KAFKA-2084_2015-06-02_17:02:00.patch, KAFKA-2084_2015-06-02_17:09:28.patch, KAFKA-2084_2015-06-02_17:10:52.patch, KAFKA-2084_2015-06-04_16:31:22.patch, KAFKA-2084_2015-06-12_10:39:35.patch, KAFKA-2084_2015-06-29_17:53:44.patch, KAFKA-2084_2015-08-04_18:50:51.patch, KAFKA-2084_2015-08-04_19:07:46.patch, KAFKA-2084_2015-08-07_11:27:51.patch, KAFKA-2084_2015-08-10_13:48:50.patch We need to be able to track the bytes-in/bytes-out rate on a per-client ID basis. This is necessary for quotas. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 33049: Patch for KAFKA-2084
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 10, 2015, 8:48 p.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description (updated) --- Signed-off-by: Aditya Auradkar aaurad...@linkedin.com Addressing Joel's comments Minor imports changes Added testcase to verify that replication traffic is not throttled Tmp commit Fixing test failure Minor Addressing Joel's comments Addressing comments Addressing comments Addressing Juns comments Minor checkstyle changes fixed test case Diffs (updated) - clients/src/main/java/org/apache/kafka/common/metrics/Quota.java d82bb0c055e631425bc1ebbc7d387baac76aeeaa clients/src/main/java/org/apache/kafka/common/metrics/QuotaViolationException.java a451e5385c9eca76b38b425e8ac856b2715fcffe clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java ca823fd4639523018311b814fde69b6177e73b97 clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java 98429da34418f7f1deba1b5e44e2e6025212edb3 clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java 544e120594de78c43581a980b1e4087b4fb98ccb clients/src/test/java/org/apache/kafka/common/utils/MockTime.java core/src/main/scala/kafka/server/ClientQuotaManager.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala 7ea509c2c41acc00430c74e025e069a833aac4e7 core/src/main/scala/kafka/server/KafkaConfig.scala dbe170f87331f43e2dc30165080d2cb7dfe5fdbf core/src/main/scala/kafka/server/KafkaServer.scala 84d4730ac634f9a5bf12a656e422fea03ad72da8 core/src/main/scala/kafka/server/ReplicaManager.scala 795220e7f63d163be90738b4c1a39687b44c1395 core/src/main/scala/kafka/server/ThrottledResponse.scala PRE-CREATION core/src/main/scala/kafka/utils/ShutdownableThread.scala fc226c863095b7761290292cd8755cd7ad0f155c core/src/test/scala/integration/kafka/api/QuotasTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala f32d206d3f52f3f9f4d649c213edd7058f4b6150 core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/33049/diff/ Testing --- Thanks, Aditya Auradkar
Re: Review Request 33049: Patch for KAFKA-2084
On Aug. 6, 2015, 2:02 a.m., Jun Rao wrote: clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java, lines 135-139 https://reviews.apache.org/r/33049/diff/18/?file=1032143#file1032143line135 Is that calculation here right? Based on the calculation in Throttler, it should be sth like the following. metricValue/quota.bound() - windowSize Also, the window size is calculated as config.sampels() * config.timeWindows(), which is inaccurate. The last window is not complete. So we need to take the current time into consideration. Finally, it seems that delayTime only makes sense for rates and not for anything else. Perhaps we can at least add a comment. Aditya Auradkar wrote: Hey Jun - Can you elaborate a little? How would we use the current time exactly? It is not clear to me how subtracting the windowSize (time unit) from a fraction (metricValue/quota.bound()) gives the right delay. I also added a comment for delayTime making sense for rates only. Aditya Auradkar wrote: I dug into the Throttler code a bit. Basically - The metricValue is the absolute actual value and not a rate. Throttler delay: (currentValueOfCounter/quotaRate) - elapsedTime Current Sensor delay: ((currentRateValue - quotaRate)/quotaRate) * elapsedTime Lets take an example (all in seconds): quotaRate = 10QPS elapedSec = 20 (Lets say 21 windows of 1 second each. The last second has not started yet) currentValueOfCounter = 250 currentRate = (250/20) = 12.5 (assuming 20 elapsed seconds. Current second may be incomplete) Throttler Formula delay = (currentValueOfCounter/quotaRate) - elapsedTime = (250/10) - 20 = 5 second delay Current Sensor delay = ((currentRateValue - quotaRate)/quotaRate) * numWindows * windowSize = ((12.5 - 10)/10) * 21 windows * 1 second window = 2.5/10 * 21 = 21/4 = 5.25 second delay I think the only discrepancy is in the elapsedTime. The last window is not complete but should be very similar because we configure many small samples. The rate calculation is done inside Rate.java which does not expose the exact elapsed time and the actual counter value. Let's examine how the rate changes because of this: The currentRate returned by the sample will still be 12.5. However, Sensor.java uses 21 as the value because we have 21 windows configured. If we got the exact elapsed time, we would use 20 elapsed seconds Potential Sensor delay = ((currentRateValue - quotaRate)/quotaRate) * elapsedTime = ((12.5 - 10)/10) * 20 = 2.5/10 * 20 = 20/4 = 5 second delay The values are now exactly identical. Given that 5.25 second delay is slightly more pessimistic, we basically throttle a bit more aggressively if the last window is not complete. We can improve this in 2 ways: - Expose elapsedTime and raw metric value from Stat.java - A better solution (IMO), is to throw QuotaViolationException from Rate.java itself. We can also add the delayTime within the Rate class since it does not make sense for other metrics. Finally, We should consider refactoring Throttler.scala to use the Rate metric for its quota computation since the formulae will be identical and only performed in 1 place. I can tackle all of this pretty quickly in a followup patch if that is acceptable. Got it. I think your calculation is correct. Perhaps we can add a comment on how the formula is derived. However, it does seem that rate includes the last partial window in calculating the rate (see Rate.measure()). So, in your formula, elapsedTime should probably be convert(now - stat.oldest(now).lastWindowMs) where now is passed in from record(). - Jun --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/#review94333 --- On Aug. 10, 2015, 8:49 p.m., Aditya Auradkar wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 10, 2015, 8:49 p.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description --- Updated patch for quotas. This patch does the following: 1. Add per-client metrics for both producer and consumers 2. Add configuration for quotas 3. Compute delay times in the metrics package and return the delay times in QuotaViolationException 4. Add a DelayQueue in KafkaApi's that can be used to throttle any type of request. Implemented request throttling for produce and fetch requests. 5. Added unit and integration test
Re: Review Request 33049: Patch for KAFKA-2084
On Aug. 6, 2015, 4:17 p.m., Jun Rao wrote: A few more comments. We need to be careful with sensors at the client-id level. Clients can come and go (e.g. console consumer). We probably don't want to hold sensors that are not longer actively used since it takes memory. So, we will need some way of removing inactive sensors. Not sure if we should add this at the metric level or at the quota level. Did you address the comment on removing inactive sensors? - Jun --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/#review94412 --- On Aug. 10, 2015, 8:49 p.m., Aditya Auradkar wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/33049/ --- (Updated Aug. 10, 2015, 8:49 p.m.) Review request for kafka, Joel Koshy and Jun Rao. Bugs: KAFKA-2084 https://issues.apache.org/jira/browse/KAFKA-2084 Repository: kafka Description --- Updated patch for quotas. This patch does the following: 1. Add per-client metrics for both producer and consumers 2. Add configuration for quotas 3. Compute delay times in the metrics package and return the delay times in QuotaViolationException 4. Add a DelayQueue in KafkaApi's that can be used to throttle any type of request. Implemented request throttling for produce and fetch requests. 5. Added unit and integration test cases for both producer and consumer 6. This doesn't include a system test. There is a separate ticket for that 7. Fixed KAFKA-2191 - (Included fix from : https://reviews.apache.org/r/34418/ ) Addressed comments from Joel and Jun Diffs - clients/src/main/java/org/apache/kafka/common/metrics/Quota.java d82bb0c055e631425bc1ebbc7d387baac76aeeaa clients/src/main/java/org/apache/kafka/common/metrics/QuotaViolationException.java a451e5385c9eca76b38b425e8ac856b2715fcffe clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java ca823fd4639523018311b814fde69b6177e73b97 clients/src/main/java/org/apache/kafka/common/metrics/stats/Rate.java 98429da34418f7f1deba1b5e44e2e6025212edb3 clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java 544e120594de78c43581a980b1e4087b4fb98ccb clients/src/test/java/org/apache/kafka/common/utils/MockTime.java core/src/main/scala/kafka/server/ClientQuotaManager.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala 7ea509c2c41acc00430c74e025e069a833aac4e7 core/src/main/scala/kafka/server/KafkaConfig.scala dbe170f87331f43e2dc30165080d2cb7dfe5fdbf core/src/main/scala/kafka/server/KafkaServer.scala 84d4730ac634f9a5bf12a656e422fea03ad72da8 core/src/main/scala/kafka/server/ReplicaManager.scala 795220e7f63d163be90738b4c1a39687b44c1395 core/src/main/scala/kafka/server/ThrottledResponse.scala PRE-CREATION core/src/main/scala/kafka/utils/ShutdownableThread.scala fc226c863095b7761290292cd8755cd7ad0f155c core/src/test/scala/integration/kafka/api/QuotasTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/ClientQuotaManagerTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala f32d206d3f52f3f9f4d649c213edd7058f4b6150 core/src/test/scala/unit/kafka/server/ThrottledResponseExpirationTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/33049/diff/ Testing --- Thanks, Aditya Auradkar
[jira] [Commented] (KAFKA-2120) Add a request timeout to NetworkClient
[ https://issues.apache.org/jira/browse/KAFKA-2120?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14681145#comment-14681145 ] Mayuresh Gharat commented on KAFKA-2120: Updated reviewboard https://reviews.apache.org/r/36858/diff/ against branch origin/trunk Add a request timeout to NetworkClient -- Key: KAFKA-2120 URL: https://issues.apache.org/jira/browse/KAFKA-2120 Project: Kafka Issue Type: New Feature Reporter: Jiangjie Qin Assignee: Mayuresh Gharat Fix For: 0.8.3 Attachments: KAFKA-2120.patch, KAFKA-2120_2015-07-27_15:31:19.patch, KAFKA-2120_2015-07-29_15:57:02.patch, KAFKA-2120_2015-08-10_19:55:18.patch Currently NetworkClient does not have a timeout setting for requests. So if no response is received for a request due to reasons such as broker is down, the request will never be completed. Request timeout will also be used as implicit timeout for some methods such as KafkaProducer.flush() and kafkaProducer.close(). KIP-19 is created for this public interface change. https://cwiki.apache.org/confluence/display/KAFKA/KIP-19+-+Add+a+request+timeout+to+NetworkClient -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 36858: Patch for KAFKA-2120
On Aug. 7, 2015, 12:36 a.m., Jun Rao wrote: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java, line 223 https://reviews.apache.org/r/36858/diff/3/?file=1024852#file1024852line223 Not sure if the test is needed. First, it seems that batch should never will be null. Second, let's say the producer can't connect to any broker. The producer can't refresh the metdata. So the leader will still be the old one and may not be null. In this case, it seems that we should still expire the records. In this case : Second, let's say the producer can't connect to any broker. The producer can't refresh the metdata. So the leader will still be the old one and may not be null. In this case, it seems that we should still expire the records., the request will eventually fail due to requestTimeout and retry exhaustion, when trying to send to broker. I was thinking on the same line of your suggestion, expiring the batch if it has exceeded the threshold even if we have metadata available, but the KIP said explicitly that Request timeout will also be used when the batches in the accumulator that are ready but not drained due to metadata missing. On Aug. 7, 2015, 12:36 a.m., Jun Rao wrote: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java, line 228 https://reviews.apache.org/r/36858/diff/3/?file=1024852#file1024852line228 We can't remove from the iterator this way. Will get a ConcurrentModificationException. Need to call iterator.remove. Ahh. My Bad. - Mayuresh --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/36858/#review94447 --- On Aug. 11, 2015, 2:55 a.m., Mayuresh Gharat wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/36858/ --- (Updated Aug. 11, 2015, 2:55 a.m.) Review request for kafka. Bugs: KAFKA-2120 https://issues.apache.org/jira/browse/KAFKA-2120 Repository: kafka Description --- Solved compile error Addressed Jason's comments for Kip-19 Addressed Jun's comments Diffs - clients/src/main/java/org/apache/kafka/clients/ClientRequest.java dc8f0f115bcda893c95d17c0a57be8d14518d034 clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java 2c421f42ed3fc5d61cf9c87a7eaa7bb23e26f63b clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java 15d00d4e484bb5d51a9ae6857ed6e024a2cc1820 clients/src/main/java/org/apache/kafka/clients/KafkaClient.java 7ab2503794ff3aab39df881bd9fbae6547827d3b clients/src/main/java/org/apache/kafka/clients/NetworkClient.java 0e51d7bd461d253f4396a5b6ca7cd391658807fa clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java d35b421a515074d964c7fccb73d260b847ea5f00 clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java ed99e9bdf7c4ea7a6d4555d4488cf8ed0b80641b clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java 9517d9d0cd480d5ba1d12f1fde7963e60528d2f8 clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java 03b8dd23df63a8d8a117f02eabcce4a2d48c44f7 clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java aa264202f2724907924985a5ecbe74afc4c6c04b clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java 4cb1e50d6c4ed55241aeaef1d3af09def5274103 clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java a152bd7697dca55609a9ec4cfe0a82c10595fbc3 clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java 06182db1c3a5da85648199b4c0c98b80ea7c6c0c clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java 0baf16e55046a2f49f6431e01d52c323c95eddf0 clients/src/main/java/org/apache/kafka/common/network/Selector.java ce20111ac434eb8c74585e9c63757bb9d60a832f clients/src/test/java/org/apache/kafka/clients/MockClient.java 9133d85342b11ba2c9888d4d2804d181831e7a8e clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 43238ceaad0322e39802b615bb805b895336a009 clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java 2c693824fa53db1e38766b8c66a0ef42ef9d0f3a clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java 5b2e4ffaeab7127648db608c179703b27b577414 clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java 8b1805d3d2bcb9fe2bacb37d870c3236aa9532c4 clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java
Re: Review Request 36858: Patch for KAFKA-2120
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/36858/ --- (Updated Aug. 11, 2015, 2:55 a.m.) Review request for kafka. Bugs: KAFKA-2120 https://issues.apache.org/jira/browse/KAFKA-2120 Repository: kafka Description (updated) --- Solved compile error Addressed Jason's comments for Kip-19 Addressed Jun's comments Diffs (updated) - clients/src/main/java/org/apache/kafka/clients/ClientRequest.java dc8f0f115bcda893c95d17c0a57be8d14518d034 clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java 2c421f42ed3fc5d61cf9c87a7eaa7bb23e26f63b clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java 15d00d4e484bb5d51a9ae6857ed6e024a2cc1820 clients/src/main/java/org/apache/kafka/clients/KafkaClient.java 7ab2503794ff3aab39df881bd9fbae6547827d3b clients/src/main/java/org/apache/kafka/clients/NetworkClient.java 0e51d7bd461d253f4396a5b6ca7cd391658807fa clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java d35b421a515074d964c7fccb73d260b847ea5f00 clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java ed99e9bdf7c4ea7a6d4555d4488cf8ed0b80641b clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java 9517d9d0cd480d5ba1d12f1fde7963e60528d2f8 clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java 03b8dd23df63a8d8a117f02eabcce4a2d48c44f7 clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java aa264202f2724907924985a5ecbe74afc4c6c04b clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java 4cb1e50d6c4ed55241aeaef1d3af09def5274103 clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java a152bd7697dca55609a9ec4cfe0a82c10595fbc3 clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordBatch.java 06182db1c3a5da85648199b4c0c98b80ea7c6c0c clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java 0baf16e55046a2f49f6431e01d52c323c95eddf0 clients/src/main/java/org/apache/kafka/common/network/Selector.java ce20111ac434eb8c74585e9c63757bb9d60a832f clients/src/test/java/org/apache/kafka/clients/MockClient.java 9133d85342b11ba2c9888d4d2804d181831e7a8e clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java 43238ceaad0322e39802b615bb805b895336a009 clients/src/test/java/org/apache/kafka/clients/producer/internals/BufferPoolTest.java 2c693824fa53db1e38766b8c66a0ef42ef9d0f3a clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java 5b2e4ffaeab7127648db608c179703b27b577414 clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java 8b1805d3d2bcb9fe2bacb37d870c3236aa9532c4 clients/src/test/java/org/apache/kafka/common/network/SelectorTest.java 158f9829ff64a969008f699e40c51e918287859e core/src/main/scala/kafka/tools/ProducerPerformance.scala 0335cc64013ffe2cdf1c4879e86e11ec8c526712 core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala ee94011894b46864614b97bbd2a98375a7d3f20b core/src/test/scala/unit/kafka/utils/TestUtils.scala eb169d8b33c27d598cc24e5a2e5f78b789fa38d3 Diff: https://reviews.apache.org/r/36858/diff/ Testing --- Thanks, Mayuresh Gharat
[jira] [Updated] (KAFKA-2120) Add a request timeout to NetworkClient
[ https://issues.apache.org/jira/browse/KAFKA-2120?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Mayuresh Gharat updated KAFKA-2120: --- Status: Patch Available (was: In Progress) Add a request timeout to NetworkClient -- Key: KAFKA-2120 URL: https://issues.apache.org/jira/browse/KAFKA-2120 Project: Kafka Issue Type: New Feature Reporter: Jiangjie Qin Assignee: Mayuresh Gharat Fix For: 0.8.3 Attachments: KAFKA-2120.patch, KAFKA-2120_2015-07-27_15:31:19.patch, KAFKA-2120_2015-07-29_15:57:02.patch, KAFKA-2120_2015-08-10_19:55:18.patch Currently NetworkClient does not have a timeout setting for requests. So if no response is received for a request due to reasons such as broker is down, the request will never be completed. Request timeout will also be used as implicit timeout for some methods such as KafkaProducer.flush() and kafkaProducer.close(). KIP-19 is created for this public interface change. https://cwiki.apache.org/confluence/display/KAFKA/KIP-19+-+Add+a+request+timeout+to+NetworkClient -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-2120) Add a request timeout to NetworkClient
[ https://issues.apache.org/jira/browse/KAFKA-2120?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Mayuresh Gharat updated KAFKA-2120: --- Attachment: KAFKA-2120_2015-08-10_19:55:18.patch Add a request timeout to NetworkClient -- Key: KAFKA-2120 URL: https://issues.apache.org/jira/browse/KAFKA-2120 Project: Kafka Issue Type: New Feature Reporter: Jiangjie Qin Assignee: Mayuresh Gharat Fix For: 0.8.3 Attachments: KAFKA-2120.patch, KAFKA-2120_2015-07-27_15:31:19.patch, KAFKA-2120_2015-07-29_15:57:02.patch, KAFKA-2120_2015-08-10_19:55:18.patch Currently NetworkClient does not have a timeout setting for requests. So if no response is received for a request due to reasons such as broker is down, the request will never be completed. Request timeout will also be used as implicit timeout for some methods such as KafkaProducer.flush() and kafkaProducer.close(). KIP-19 is created for this public interface change. https://cwiki.apache.org/confluence/display/KAFKA/KIP-19+-+Add+a+request+timeout+to+NetworkClient -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-1683) Implement a session concept in the socket server
[ https://issues.apache.org/jira/browse/KAFKA-1683?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14681149#comment-14681149 ] Eugene Miretsky commented on KAFKA-1683: Would this patch include the ability to authorizer as different users? Or will it be handled in another JIRA? Implement a session concept in the socket server -- Key: KAFKA-1683 URL: https://issues.apache.org/jira/browse/KAFKA-1683 Project: Kafka Issue Type: Sub-task Components: security Affects Versions: 0.9.0 Reporter: Jay Kreps Assignee: Gwen Shapira Fix For: 0.8.3 Attachments: KAFKA-1683.patch, KAFKA-1683.patch To implement authentication we need a way to keep track of some things between requests. The initial use for this would be remembering the authenticated user/principle info, but likely more uses would come up (for example we will also need to remember whether and which encryption or integrity measures are in place on the socket so we can wrap and unwrap writes and reads). I was thinking we could just add a Session object that might have a user field. The session object would need to get added to RequestChannel.Request so it is passed down to the API layer with each request. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 35437: Patch for KAFKA-2202
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/35437/#review94836 --- Ship it! LGTM. Verified it gets rid of KAFKA-1828 too. - Ewen Cheslack-Postava On June 14, 2015, 11:27 a.m., Manikumar Reddy O wrote: --- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/35437/ --- (Updated June 14, 2015, 11:27 a.m.) Review request for kafka. Bugs: KAFKA-2202 https://issues.apache.org/jira/browse/KAFKA-2202 Repository: kafka Description --- while computing stats, consumerTimeoutMs is considered only during ConsumerTimeout exception senarios; This should solve KAFKA-1828 also Diffs - core/src/main/scala/kafka/tools/ConsumerPerformance.scala 903318d15893af08104a97499798c9ad0ba98013 Diff: https://reviews.apache.org/r/35437/diff/ Testing --- Thanks, Manikumar Reddy O
[jira] [Updated] (KAFKA-2338) Warn users if they change max.message.bytes that they also need to update broker and consumer settings
[ https://issues.apache.org/jira/browse/KAFKA-2338?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jun Rao updated KAFKA-2338: --- Reviewer: Gwen Shapira (was: Jun Rao) Warn users if they change max.message.bytes that they also need to update broker and consumer settings -- Key: KAFKA-2338 URL: https://issues.apache.org/jira/browse/KAFKA-2338 Project: Kafka Issue Type: Bug Components: core Affects Versions: 0.8.2.1 Reporter: Ewen Cheslack-Postava Assignee: Edward Ribeiro Fix For: 0.8.3 Attachments: KAFKA-2338.patch, KAFKA-2338_2015-07-18_00:37:31.patch, KAFKA-2338_2015-07-21_13:21:19.patch We already have KAFKA-1756 filed to more completely address this issue, but it is waiting for some other major changes to configs to completely protect users from this problem. This JIRA should address the low hanging fruit to at least warn users of the potential problems. Currently the only warning is in our documentation. 1. Generate a warning in the kafka-topics.sh tool when they change this setting on a topic to be larger than the default. This needs to be very obvious in the output. 2. Currently, the broker's replica fetcher isn't logging any useful error messages when replication can't succeed because a message size is too large. Logging an error here would allow users that get into a bad state to find out why it is happening more easily. (Consumers should already be logging a useful error message.) -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 34492: Patch for KAFKA-2210
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/common/AuthorizationException.scala, line 24
https://reviews.apache.org/r/34492/diff/9/?file=1018318#file1018318line24
Exceptions without a message are discouraged, so I would remove the
no-args constructor.
fixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/Authorizer.scala, line 64
https://reviews.apache.org/r/34492/diff/9/?file=1018321#file1018321line64
Should this be called `removeResource` instead? Good to avoid method
overloads when possible.
its actually not removing the resource itself though, it is only removing acls
attached to the acls. Want to avoid naming it somehting that will be misleading.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/KafkaPrincipal.scala, line 26
https://reviews.apache.org/r/34492/diff/9/?file=1018322#file1018322line26
Type annotations are usually not used for local vals.
done for all classes under auth package.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/KafkaPrincipal.scala, line 28
https://reviews.apache.org/r/34492/diff/9/?file=1018322#file1018322line28
Code contention: no braces for single expression.
again done for all classed under auth package.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/KafkaPrincipal.scala, line 32
https://reviews.apache.org/r/34492/diff/9/?file=1018322#file1018322line32
This could be written in a nicer way like this:
```
str.split(Separator, 2) match {
case Array(principalType, name, _*) = new
KafkaPrincipal(principalType, name)
case s = throw IllegalArgumentException(...)
}
```
made changes at KafkaPrincipal and Resource.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/KafkaPrincipal.scala, line 41
https://reviews.apache.org/r/34492/diff/9/?file=1018322#file1018322line41
If you make this a case class, you get decent `toString`, `equals` and
`hashCode` by default. Also, the `val` is then not needed for the fields.
I have changed all the classes to case class. This basically means all the
equalities are now case sensitive.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/Operation.scala, line 38
https://reviews.apache.org/r/34492/diff/9/?file=1018323#file1018323line38
`find` is better than `filter` here as it stops once the match is
found. Also, `headOption` is not needed then.
fixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/Operation.scala, line 42
https://reviews.apache.org/r/34492/diff/9/?file=1018323#file1018323line42
Code convention: no braces needed for single expression.
Also, no need for `()` since there is no side-effect here.
fixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/PermissionType.scala, line 38
https://reviews.apache.org/r/34492/diff/9/?file=1018324#file1018324line38
Same points as the ones in `Operation`.
fixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/Resource.scala, line 22
https://reviews.apache.org/r/34492/diff/9/?file=1018325#file1018325line22
Space after `,`.
fixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/Resource.scala, line 31
https://reviews.apache.org/r/34492/diff/9/?file=1018325#file1018325line31
Same comments as in `KafkaPrincipal.fromString`.
fixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/Resource.scala, line 41
https://reviews.apache.org/r/34492/diff/9/?file=1018325#file1018325line41
Make it a case class?
mixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/security/auth/ResourceType.scala, line 52
https://reviews.apache.org/r/34492/diff/9/?file=1018326#file1018326line52
Same comments as in `Operation`.
fixed.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/test/scala/unit/kafka/security/auth/AclTest.scala, line 44
https://reviews.apache.org/r/34492/diff/9/?file=1018330#file1018330line44
Type annotation is generally not needed for local `vals` (there are a
number of instances of this).
Fixed all instances that I could find.
On July 28, 2015, 5:18 p.m., Ismael Juma wrote:
core/src/main/scala/kafka/server/KafkaApis.scala, line 104
https://reviews.apache.org/r/34492/diff/9/?file=1018327#file1018327line104
In general, `Option.get` should be avoided. Instead of manually
checking if it's defined, use safer methods. For example:
```authorizer.foreach { a =
if (!a.authorize(...)
Re: Review Request 34492: Patch for KAFKA-2210
On Julho 21, 2015, 1:30 a.m., Edward Ribeiro wrote:
core/src/main/scala/kafka/security/auth/Acl.scala, line 71
https://reviews.apache.org/r/34492/diff/8/?file=1017296#file1017296line71
Disclaimer: I am not claiming that you should change the code commented
here.
Okay, for getting rid of the dreaded
``collection.mutable.HashSet[Acl]()``, you have two options, afaik:
1. use ``(for (i - list) yield i).toSet``. In the current code it
would be something like:
```
val acls = (for (item - aclSet) {
val principals: List[KafkaPrincipal] =
item(PrincipalKey).asInstanceOf[List[String]].map(principal =
KafkaPrincipal.fromString(principal))
val permissionType: PermissionType =
PermissionType.fromString(item(PermissionTypeKey).asInstanceOf[String])
val operations: List[Operation] =
item(OperationKey).asInstanceOf[List[String]].map(operation =
Operation.fromString(operation))
val hosts: List[String] = item(HostsKey).asInstanceOf[List[String]]
yield new Acl(principals.toSet, permissionType, hosts.toSet,
operations.toSet)
}).toSet
```
The surrounding parenthesis around the ``for`` comprehesion are
important as ``yield`` would return the same Collection type from aclSet (a
List in this case).
2. To use a (private) helper recursive function like, for example:
```
private def listToSet(list: List[Map[String, Any]]): Set[Acl] = {
list match {
case item::tail = {
// L#72 - L#75 processing over `item`
Set(new Acl(...)) ++ listToSet(tail)
}
case Nil = Set.empty[Acl]
}
}
```
can call it from ``fromJson`` on ``aclSet`` instead of doing a
``foreach``.
In fact, most of lines L#72 - L#75 are composed of Lists that
eventually get converted to set (principals, hosts, operations and acls),
so you could generify the helper function above, so that you could pass a
'convertion' function, but here I am wary of the complexity of the code
starting to outweight the benefits (?) of not using mutable data
structures... Nevertheless, it would look like:
```
def listToSet[T,K](list: List[T], f: T = K): Set[K] = {
list match {
case head::tail = Set(f(head)) ++ listToSet(tail, f)
case Nil = Set.empty[K]
}
}
```
Parth Brahmbhatt wrote:
I haven't changed it for now dont really think to .toSet will be that bad.
Agree! All the other options leave the code more obfuscated than it needs to
be, imo. Only if there was some strict code guideline to use pure Functional
Programming we would need to resort to one of those options I described.
- Edward
---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/34492/#review92345
---
On Ago. 11, 2015, 1:32 a.m., Parth Brahmbhatt wrote:
---
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/34492/
---
(Updated Ago. 11, 2015, 1:32 a.m.)
Review request for kafka.
Bugs: KAFKA-2210
https://issues.apache.org/jira/browse/KAFKA-2210
Repository: kafka
Description
---
Addressing review comments from Jun.
Adding CREATE check for offset topic only if the topic does not exist already.
Addressing some more comments.
Removing acl.json file
Moving PermissionType to trait instead of enum. Following the convention for
defining constants.
Adding authorizer.config.path back.
Addressing more comments from Jun.
Addressing more comments.
Now addressing Ismael's comments. Case sensitive checks.
Diffs
-
core/src/main/scala/kafka/api/OffsetRequest.scala
f418868046f7c99aefdccd9956541a0cb72b1500
core/src/main/scala/kafka/common/AuthorizationException.scala PRE-CREATION
core/src/main/scala/kafka/common/ErrorMapping.scala
c75c68589681b2c9d6eba2b440ce5e58cddf6370
core/src/main/scala/kafka/network/RequestChannel.scala
20741281dcaa76374ea6f86a2185dad27b515339
core/src/main/scala/kafka/security/auth/Acl.scala PRE-CREATION
core/src/main/scala/kafka/security/auth/Authorizer.scala PRE-CREATION
core/src/main/scala/kafka/security/auth/KafkaPrincipal.scala PRE-CREATION
core/src/main/scala/kafka/security/auth/Operation.scala PRE-CREATION
core/src/main/scala/kafka/security/auth/PermissionType.scala PRE-CREATION
core/src/main/scala/kafka/security/auth/Resource.scala PRE-CREATION
Re: [DISCUSS] KIP-28 - Add a transform client for data processing
Hello folks, I have updated the KIP page with some detailed API / architecture / packaging proposals, along with the long promised first patch in PR: https://cwiki.apache.org/confluence/display/KAFKA/KIP-28+-+Add+a+processor+client https://github.com/apache/kafka/pull/130 Any feedbacks / comments are more than welcomed. Guozhang On Mon, Aug 10, 2015 at 6:55 PM, Guozhang Wang wangg...@gmail.com wrote: Hi Jun, 1. I have removed the streamTime in punctuate() since it is not only triggered by clock time, detailed explanation can be found here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-28+-+Add+a+processor+client#KIP-28-Addaprocessorclient-StreamTime 2. Yes, if users do not schedule a task, then punctuate will never fire. 3. Yes, I agree. The reason it was implemented in this way is that the state store registration call is triggered by the users. However I think it is doable to change that API so that it will be more natural to have sth. like: context.createStore(store-name, store-type). Guozhang On Tue, Aug 4, 2015 at 9:17 AM, Jun Rao j...@confluent.io wrote: A few questions/comments. 1. What's streamTime passed to punctuate()? Is that just the current time? 2. Is punctuate() only called if schedule() is called? 3. The way the KeyValueStore is created seems a bit weird. Since this is part of the internal state managed by KafkaProcessorContext, it seems there should be an api to create the KeyValueStore from KafkaProcessorContext, instead of passing context to the constructor of KeyValueStore? Thanks, Jun On Thu, Jul 23, 2015 at 5:59 PM, Guozhang Wang wangg...@gmail.com wrote: Hi all, I just posted KIP-28: Add a transform client for data processing https://cwiki.apache.org/confluence/display/KAFKA/KIP-28+-+Add+a+transform+client+for+data+processing . The wiki page does not yet have the full design / implementation details, and this email is to kick-off the conversation on whether we should add this new client with the described motivations, and if yes what features / functionalities should be included. Looking forward to your feedback! -- Guozhang -- -- Guozhang -- -- Guozhang
[jira] [Commented] (KAFKA-2210) KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation.
[ https://issues.apache.org/jira/browse/KAFKA-2210?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14681075#comment-14681075 ] Parth Brahmbhatt commented on KAFKA-2210: - Updated reviewboard https://reviews.apache.org/r/34492/diff/ against branch origin/trunk KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation. -- Key: KAFKA-2210 URL: https://issues.apache.org/jira/browse/KAFKA-2210 Project: Kafka Issue Type: Sub-task Components: security Reporter: Parth Brahmbhatt Assignee: Parth Brahmbhatt Fix For: 0.8.3 Attachments: KAFKA-2210.patch, KAFKA-2210_2015-06-03_16:36:11.patch, KAFKA-2210_2015-06-04_16:07:39.patch, KAFKA-2210_2015-07-09_18:00:34.patch, KAFKA-2210_2015-07-14_10:02:19.patch, KAFKA-2210_2015-07-14_14:13:19.patch, KAFKA-2210_2015-07-20_16:42:18.patch, KAFKA-2210_2015-07-21_17:08:21.patch, KAFKA-2210_2015-08-10_18:31:54.patch This is the first subtask for Kafka-1688. As Part of this jira we intend to agree on all the public entities, configs and changes to existing kafka classes to allow pluggable authorizer implementation. Please see KIP-11 https://cwiki.apache.org/confluence/display/KAFKA/KIP-11+-+Authorization+Interface for detailed design. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-2210) KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation.
[ https://issues.apache.org/jira/browse/KAFKA-2210?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Parth Brahmbhatt updated KAFKA-2210: Attachment: KAFKA-2210_2015-08-10_18:31:54.patch KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation. -- Key: KAFKA-2210 URL: https://issues.apache.org/jira/browse/KAFKA-2210 Project: Kafka Issue Type: Sub-task Components: security Reporter: Parth Brahmbhatt Assignee: Parth Brahmbhatt Fix For: 0.8.3 Attachments: KAFKA-2210.patch, KAFKA-2210_2015-06-03_16:36:11.patch, KAFKA-2210_2015-06-04_16:07:39.patch, KAFKA-2210_2015-07-09_18:00:34.patch, KAFKA-2210_2015-07-14_10:02:19.patch, KAFKA-2210_2015-07-14_14:13:19.patch, KAFKA-2210_2015-07-20_16:42:18.patch, KAFKA-2210_2015-07-21_17:08:21.patch, KAFKA-2210_2015-08-10_18:31:54.patch This is the first subtask for Kafka-1688. As Part of this jira we intend to agree on all the public entities, configs and changes to existing kafka classes to allow pluggable authorizer implementation. Please see KIP-11 https://cwiki.apache.org/confluence/display/KAFKA/KIP-11+-+Authorization+Interface for detailed design. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: Review Request 34492: Patch for KAFKA-2210
--- This is an automatically generated e-mail. To reply, visit: https://reviews.apache.org/r/34492/ --- (Updated Aug. 11, 2015, 1:32 a.m.) Review request for kafka. Bugs: KAFKA-2210 https://issues.apache.org/jira/browse/KAFKA-2210 Repository: kafka Description (updated) --- Addressing review comments from Jun. Adding CREATE check for offset topic only if the topic does not exist already. Addressing some more comments. Removing acl.json file Moving PermissionType to trait instead of enum. Following the convention for defining constants. Adding authorizer.config.path back. Addressing more comments from Jun. Addressing more comments. Now addressing Ismael's comments. Case sensitive checks. Diffs (updated) - core/src/main/scala/kafka/api/OffsetRequest.scala f418868046f7c99aefdccd9956541a0cb72b1500 core/src/main/scala/kafka/common/AuthorizationException.scala PRE-CREATION core/src/main/scala/kafka/common/ErrorMapping.scala c75c68589681b2c9d6eba2b440ce5e58cddf6370 core/src/main/scala/kafka/network/RequestChannel.scala 20741281dcaa76374ea6f86a2185dad27b515339 core/src/main/scala/kafka/security/auth/Acl.scala PRE-CREATION core/src/main/scala/kafka/security/auth/Authorizer.scala PRE-CREATION core/src/main/scala/kafka/security/auth/KafkaPrincipal.scala PRE-CREATION core/src/main/scala/kafka/security/auth/Operation.scala PRE-CREATION core/src/main/scala/kafka/security/auth/PermissionType.scala PRE-CREATION core/src/main/scala/kafka/security/auth/Resource.scala PRE-CREATION core/src/main/scala/kafka/security/auth/ResourceType.scala PRE-CREATION core/src/main/scala/kafka/server/KafkaApis.scala 7ea509c2c41acc00430c74e025e069a833aac4e7 core/src/main/scala/kafka/server/KafkaConfig.scala dbe170f87331f43e2dc30165080d2cb7dfe5fdbf core/src/main/scala/kafka/server/KafkaServer.scala 84d4730ac634f9a5bf12a656e422fea03ad72da8 core/src/test/scala/unit/kafka/security/auth/AclTest.scala PRE-CREATION core/src/test/scala/unit/kafka/security/auth/KafkaPrincipalTest.scala PRE-CREATION core/src/test/scala/unit/kafka/security/auth/OperationTest.scala PRE-CREATION core/src/test/scala/unit/kafka/security/auth/PermissionTypeTest.scala PRE-CREATION core/src/test/scala/unit/kafka/security/auth/ResourceTypeTest.scala PRE-CREATION core/src/test/scala/unit/kafka/server/KafkaConfigConfigDefTest.scala PRE-CREATION Diff: https://reviews.apache.org/r/34492/diff/ Testing --- Thanks, Parth Brahmbhatt
Re: [jira] [Commented] (KAFKA-1690) new java producer needs ssl support as a client
Thanks for the details. I ran the same test from the branch you used with oracle java 8 , java 7 and IBM java 8 on linux in a loop of 50 didn't see any issue. Can you give me details on which version of IBM jdk did you used and what OS. java -version java version 1.8.0 Java(TM) SE Runtime Environment (build pxi3280sr1fp10- 20150711_01(SR1 FP10)) IBM J9 VM (build 2.8, JRE 1.8.0 Linux x86-32 20150630_255633 (JIT enabled, AOT enabled) J9VM - R28_jvm.28_20150630_1742_B255633 JIT - tr.r14.java_20150625_95081.01 GC - R28_jvm.28_20150630_1742_B255633 J9CL - 20150630_255633) JCL - 20150711_01 based on Oracle jdk8u51-b15 Thanks, Harsha On Mon, Aug 10, 2015, at 01:52 PM, Rajini Sivaram wrote: Harsha, I am using the code from https://github.com/harshach/kafka/tree/KAFKA-1690-V1 with the latest commit on 25 July which I think corresponds to the latest patch in KAFKA- 1690. Is that correct? I have run SSLConsumerTest, SSLProducerSendTest and SSLSelectorTest several times and the only one that fails to complete is SSLSelectorTest.testRenegotiate(). I added some trace to the code and the sequence in the failed run is: 1. Handshake starts 2. Handshake finishes with handshakeStatus=FINISHED appReadBuffer=java.nio.DirectByteBuffer[pos=0 lim=16916 cap=16916] *(there is no data in appReadBuffer)* 3. Several reads and writes occur 4. read() results in handshake() due to renegotiation. appReadBuffer=java.nio.DirectByteBuffer[pos=0 lim=16916 cap=16916] *(no data in appReadBuffer)* 5. Handshake status changes from NEED_TASK to NEED_WRAP to NEED_UNWRAP 6. Unwrap call during handshake results in data in appReadBuffer: handshakeStatus=NEED_UNWRAP appReadBuffer=java.nio.DirectByteBuffer[pos=100 lim=16916 cap=16916] *(data is now in appReadBuffer)* 7. Handshake status changes to NEED_UNWRAP followed by FINISHED. Final call to handshake() returns with handshakeStatus=NOT_HANDSHAKING appReadBuffer=java.nio.DirectByteBuffer[pos=100 lim=16916 cap=16916] *(data is still in appReadBuffer)* Test continues to call selector.poll() forever. But no more data arrives on the channel since all the data in the test has already arrived, and hence SSLTransportLayer.read() is never invoked on the channel and the data that was unwrapped earlier simply remains in appReadBuffer. This is not a scenario that occurs in the other tests where some data always arrives on the network after the handshake() resulting in SSLTransportLayer.read() being invoked. I can recreate this problem quite easily with IBM JDK (it hangs around one in ten runs), so if you require more trace, please let me know. Thank you, Rajini On Mon, Aug 10, 2015 at 5:29 PM, Sriharsha Chintalapani harsh...@fastmail.fm wrote: Thanks for testing out Rajini. I did ran that test in a loop and it never hanged for me. I am hoping you are using the latest patch since the data left over issue is addressed in latest patch. Also if thats an issue SSLConsumerTest and SSLProducerTest will hang too. Did you notice those are having any issues? I am addressing left over reviews will be sending a new patch in a day or two. Thanks, Harsha On August 10, 2015 at 3:35:34 AM, Rajini Sivaram (rajinisiva...@googlemail.com) wrote: I was running a Kafka cluster with the latest SSL patch over the weekend with IBM JRE, and it has been running fine without any issues. There was light load on the cluster throughout and intermittent heavy load, all using SSL clients. However I am seeing an intermittent unit test hang in org.apache.kafka.common.network.SSLSelectorTest.testRenegotiate(). I am not sure if that is related to the IBM JRE I am using for the build. It looks like data left in appReadBuffer after a handshake may not get processed if no more data arrives on the network, causing the test to loop forever. It will be good if this can be fixed (or at least the test commented out) before the code is committed to avoid breaking the build. Even though there are quite a few outstanding review comments, I do agree that being such a big patch, it will be good to commit the code soon and work on minor issues afterwards. On Mon, Aug 10, 2015 at 12:19 AM, Jun Rao (JIRA) j...@apache.org wrote: [ https://issues.apache.org/jira/browse/KAFKA-1690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14679397#comment-14679397 ] Jun Rao commented on KAFKA-1690: [~rsivaram], this jira is getting pretty close to be committed. Could you test this out on an IBM jvm to see if there is any issue especially with respect to the usage of the sendfile api? new java producer needs ssl support as a client --- Key: KAFKA-1690 URL:
[GitHub] kafka pull request: KIP-28: First patch
GitHub user guozhangwang opened a pull request:
https://github.com/apache/kafka/pull/130
KIP-28: First patch
Some open questions collected so far on the first patch. Thanks @gwenshap
@jkreps.
1. Can we hide the Chooser interface from users? In other words, if users
can specify the time on each fetched messages from Kafka, would a hard-coded
MinTimestampMessageChooser be sufficient so that we can move TimestampTracker /
RecordQueue / Chooser / RecordCollector / etc all to the internal folders?
2. Shall we split the o.a.k.clients into two folders, with
o.a.k.clients.processor in stream? Or should we just remove
o.a.k.clients.processor and make everything under o.a.k.stream? In addition,
currently there is a cyclic dependency between that two, would better to break
it in the end state.
3. Topology API: requiring users to instantiate their own Topology class
with the overridden build() function is a little awkward. Instead it would be
great to let users explicitly build the topology in Main and pass it in as a
class:
```
Topology myTopology = new TopologyBuilder(defaultDeser)
.addProcessor(my-processor, MyProcessor.class, new Source(my-source))
.addProcessor(my-other-processor, MyOtherProcessor.class, my-processor);
KafkaStreaming streaming = new KafkaStreaming(config, myTopology);
streaming.run();
```
So the implementation of KStream.filter look instead like this:
```
public KStreamK, V filter(PredicateK, V predicate) {
KStreamFilterK, V filter = new KStreamFilter();
topology.addProcessor(KStreamFilter.class, new Configs(predicate,
predicate));
return this;
}
```
The advantage is that the user code can now get rid of the whole Topology
class with the builder. I think the order of execution for that API is quite
unintuitive.
4. We can probably move the forward() function from Processor to
ProcessorContext, and split ProcessorContext into two classes, one with all the
function calls as commit / send / schedule / forward, and another with the
metadata function calls as topic / partition / offset / timestamp.
5. Merge ProcessorConfigs with ProcessorProperties.
6. Consider moving the external dependencies such as RocksDB into a
separate jar? For example we can just include a kafka-stream-rocksdb.jar which
includes the RocksDBKeyValueStore only, and later on when we deprecate / remove
such implementations we can simply remove the jar itself.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/confluentinc/kafka streaming
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/kafka/pull/130.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 #130
commit 1527f277dc33a18ce348a357d7883349af72fc49
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-24T16:40:46Z
First commit
commit abc220a0f2217a69103466fc3e9bdcf92502a15a
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-25T17:56:58Z
stream synchronization
commit dc200a460f31331bbb5b2bcc3f0567e5fb80904e
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-29T16:27:06Z
ported many stuff from Jay's streming prototype
commit 34d02b21e46902df77e844b1b8b30043fa98cff3
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-29T20:52:49Z
removed punctuate method, added punctuationqueue per streamsynchronizer
commit d7c068990513c4ded6f0efbd04d9995c1a69db85
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-29T21:39:13Z
fixed compile warnings
commit f81d463df655349af25889e2a3319403fa017d6d
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-29T23:19:49Z
added sync to PunctuationSchedulerImpl
commit 480ee6d41e26499a3e035f8471a82431b3733014
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-30T00:09:23Z
pass stream time to puctuate()
commit b20bbd2d596f6e5e53a0eaab0f8f88275c5e8e8b
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-30T16:46:16Z
removed flush method from KStream and Processor
commit 583348bad6e7dd2ea62078231d757de72a7ea0ec
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-30T17:09:35Z
simplified recordqueue interface
commit 8cce08db01c0d3df62ba52316fade3dfd58cba24
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-30T18:03:40Z
separated timestamp stacking from queue impl
commit 31576183c1011e68fe54774d4160c81ab17c59ab
Author: Yasuhiro Matsuda yasuhiro.mats...@gmail.com
Date: 2015-06-30T19:03:18Z
comments
commit
Re: [DISCUSS] KIP-28 - Add a transform client for data processing
Hi Jiangjie, Not sure I understand the What If user have interleaved groups of messages, each group makes a complete logic? Could you elaborate a bit? About the committing functionality, it currently will only commit up to the processed message's offset; the commit() call it self actually does more than consumer committing offsets, but together with flushing the local state and the producer. Guozhang On Fri, Jul 31, 2015 at 9:20 PM, Jiangjie Qin j...@linkedin.com.invalid wrote: I think the abstraction of processor would be useful. It is not quite clear to me yet though which grid in the following API analysis chart this processor is trying to satisfy. https://cwiki.apache.org/confluence/display/KAFKA/New+consumer+API+change+proposal For example, in current proposal. It looks user will only be able to commit offsets for the last seen message. What If user have interleaved groups of messages, each group makes a complete logic? In that case, user will not have a safe boundary to commit offset. Is the processor client only intended to address the static topic data stream with semi-auto offset commit (which means user can only commit the last seen message)? Jiangjie (Becket) Qin On Thu, Jul 30, 2015 at 2:32 PM, James Cheng jch...@tivo.com wrote: I agree with Sriram and Martin. Kafka is already about providing streams of data, and so Kafka Streams or anything like that is confusing to me. This new library is about making it easier to process the data. -James On Jul 30, 2015, at 9:38 AM, Aditya Auradkar aaurad...@linkedin.com.INVALID wrote: Personally, I prefer KafkaStreams just because it sounds nicer. For the reasons identified above, KafkaProcessor or KProcessor is more apt but sounds less catchy (IMO). I also think we should prefix with Kafka (rather than K) because we will then have 3 clients: KafkaProducer, KafkaConsumer and KafkaProcessor which is very nice and consistent. Aditya On Thu, Jul 30, 2015 at 9:17 AM, Gwen Shapira gshap...@cloudera.com wrote: I think its also a matter of intent. If we see it as yet another client library, than Processor (to match Producer and Consumer) will work great. If we see it is a stream processing framework, the name has to start with S to follow existing convention. Speaking of naming conventions: You know how people have stack names for technologies that are usually used in tandem? ELK, LAMP, etc. The pattern of Kafka - Stream Processor - NoSQL Store is super common. KSN stack doesn't sound right, though. Maybe while we are bikeshedding, someone has ideas in that direction :) On Thu, Jul 30, 2015 at 2:01 AM, Sriram Subramanian srsubraman...@linkedin.com.invalid wrote: I had the same thought. Kafka processor, KProcessor or even Kafka stream processor is more relevant. On Jul 30, 2015, at 2:09 PM, Martin Kleppmann mar...@kleppmann.com wrote: I'm with Sriram -- Kafka is all about streams already (or topics, to be precise, but we're calling it stream processing not topic processing), so I find Kafka Streams, KStream and Kafka Streaming all confusing, since they seem to imply that other bits of Kafka are not about streams. I would prefer The Processor API or Kafka Processors or Kafka Processing Client or KProcessor, or something along those lines. On 30 Jul 2015, at 15:07, Guozhang Wang wangg...@gmail.com wrote: I would vote for KStream as it sounds sexier (is it only me??), second to that would be Kafka Streaming. On Wed, Jul 29, 2015 at 6:08 PM, Jay Kreps j...@confluent.io wrote: Also, the most important part of any prototype, we should have a name for this producing-consumer-thingamgigy: Various ideas: - Kafka Streams - KStream - Kafka Streaming - The Processor API - Metamorphosis - Transformer API - Verwandlung For my part I think what people are trying to do is stream processing with Kafka so I think something that evokes Kafka and stream processing is preferable. I like Kafka Streams or Kafka Streaming followed by KStream. Transformer kind of makes me think of the shape-shifting cars. Metamorphosis is cool and hilarious but since we are kind of envisioning this as more limited scope thing rather than a massive framework in its own right I actually think it should have a descriptive name rather than a personality of it's own. Anyhow let the bikeshedding commence. -Jay On Thu, Jul 23, 2015 at 5:59 PM, Guozhang Wang wangg...@gmail.com wrote: Hi all, I just posted KIP-28: Add a transform client for data processing https://cwiki.apache.org/confluence/display/KAFKA/KIP-28+-+Add+a+transform+client+for+data+processing . The wiki page does not yet have the full design / implementation details, and this email is
Re: [DISCUSS] KIP-28 - Add a transform client for data processing
Hi Jun, 1. I have removed the streamTime in punctuate() since it is not only triggered by clock time, detailed explanation can be found here: https://cwiki.apache.org/confluence/display/KAFKA/KIP-28+-+Add+a+processor+client#KIP-28-Addaprocessorclient-StreamTime 2. Yes, if users do not schedule a task, then punctuate will never fire. 3. Yes, I agree. The reason it was implemented in this way is that the state store registration call is triggered by the users. However I think it is doable to change that API so that it will be more natural to have sth. like: context.createStore(store-name, store-type). Guozhang On Tue, Aug 4, 2015 at 9:17 AM, Jun Rao j...@confluent.io wrote: A few questions/comments. 1. What's streamTime passed to punctuate()? Is that just the current time? 2. Is punctuate() only called if schedule() is called? 3. The way the KeyValueStore is created seems a bit weird. Since this is part of the internal state managed by KafkaProcessorContext, it seems there should be an api to create the KeyValueStore from KafkaProcessorContext, instead of passing context to the constructor of KeyValueStore? Thanks, Jun On Thu, Jul 23, 2015 at 5:59 PM, Guozhang Wang wangg...@gmail.com wrote: Hi all, I just posted KIP-28: Add a transform client for data processing https://cwiki.apache.org/confluence/display/KAFKA/KIP-28+-+Add+a+transform+client+for+data+processing . The wiki page does not yet have the full design / implementation details, and this email is to kick-off the conversation on whether we should add this new client with the described motivations, and if yes what features / functionalities should be included. Looking forward to your feedback! -- Guozhang -- -- Guozhang
[jira] [Commented] (KAFKA-2411) remove usage of BlockingChannel in the broker
[ https://issues.apache.org/jira/browse/KAFKA-2411?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680045#comment-14680045 ] ASF GitHub Bot commented on KAFKA-2411: --- GitHub user ijuma opened a pull request: https://github.com/apache/kafka/pull/127 KAFKA-2411; [WIP] remove usage of blocking channel This PR builds on the work from @harshach and only the last commit is relevant. Opening the PR for getting feedback. You can merge this pull request into a Git repository by running: $ git pull https://github.com/ijuma/kafka kafka-2411-remove-usage-of-blocking-channel Alternatively you can review and apply these changes as the patch at: https://github.com/apache/kafka/pull/127.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 #127 commit 8ca558920347733ddf7a924463c93620e976a3f3 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-04-28T07:29:53Z KAFKA-1690. new java producer needs ssl support as a client. commit 754a121e7582f1452a9ae3a3ab72c58cf284da1d Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T06:02:01Z KAFKA-1690. new java producer needs ssl support as a client. commit 98a90ae9d80ea8f5ab4780569d1c4e301dd16c4e Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T06:18:13Z KAFKA-1690. new java producer needs ssl support as a client. commit 804da7a015be2f98a1bb867ee5d42aa8009a37dd Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T06:31:25Z KAFKA-1690. new java producer needs ssl support as a client. commit ee16e8e6f92ac2baf0e41d3019b7f8aef39b1506 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T23:09:01Z KAFKA-1690. new java producer needs ssl support as a client. SSLFactory tests. commit 2dd826be4a6ebe7064cb19ff21fe23950a1bafc2 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-12T23:09:38Z KAFKA-1690. new java producer needs ssl support as a client. Added PrincipalBuilder. commit 2cddad80f6a4a961b6932879448e532dab4e637e Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-15T14:17:37Z KAFKA-1690. new java producer needs ssl support as a client. Addressing reviews. commit ca0456dc01def337ee1711cabd9c4e9df4af61ee Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-20T21:23:29Z KAFKA-1690. new java producer needs ssl support as a client. Addressing reviews. commit 7e3a4cfc58932aab4288677111af52f94c9012b6 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-20T21:37:52Z KAFKA-1690. new java producer needs ssl support as a client. Addressing reviews. commit 9bdceb24f8682184f7fb39578f239a7b6dde Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-21T16:50:52Z KAFKA-1690. new java producer needs ssl support as a client. Fixed minor issues with the patch. commit 65396b5cabeaf61579c6e6422848877fc7a896a9 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-21T17:27:11Z KAFKA-1690. new java producer needs ssl support as a client. Fixed minor issues with the patch. commit b37330a7b4ec3adfba4f0c6e33ab172be03406be Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-29T03:57:06Z KAFKA-1690. new java producer needs ssl support as a client. commit fe595fd4fda45ebd7c5da88ee093ab17817bb94d Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-04T01:43:34Z KAFKA-1690. new java producer needs ssl support as a client. commit 247264ce35a04d14b97c87dcb88378ad1dbe0986 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-08T16:07:14Z Merge remote-tracking branch 'refs/remotes/origin/trunk' into KAFKA-1690-V1 commit 050782b9f47f4c61b22ef065ec4798ccbdb962d3 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-16T15:46:05Z KAFKA-1690. Broker side ssl changes. commit 9328ffa464711a835be8935cb09922230e0e1a58 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-20T17:47:01Z KAFKA-1684. SSL for socketServer. commit eda92cb5f9d2ae749903eac5453a6fdb49685964 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T03:01:30Z KAFKA-1690. Added SSLProducerSendTest and fixes to get right port for SSL. commit f10e28b2f2b10d91db9c1aba977fc578b4c4c633 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T03:47:54Z Merge branch 'trunk' into KAFKA-1690-V1 commit f60c95273b3b814792d0da9264a75939049dcc5f Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T04:45:58Z KAFKA-1690. Post merge fixes. commit 8f7ba892502b09cb7cc05d75270352815fb1c42c Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T22:35:52Z KAFKA-1690. Added
[GitHub] kafka pull request: KAFKA-2411; [WIP] remove usage of blocking cha...
GitHub user ijuma opened a pull request: https://github.com/apache/kafka/pull/127 KAFKA-2411; [WIP] remove usage of blocking channel This PR builds on the work from @harshach and only the last commit is relevant. Opening the PR for getting feedback. You can merge this pull request into a Git repository by running: $ git pull https://github.com/ijuma/kafka kafka-2411-remove-usage-of-blocking-channel Alternatively you can review and apply these changes as the patch at: https://github.com/apache/kafka/pull/127.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 #127 commit 8ca558920347733ddf7a924463c93620e976a3f3 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-04-28T07:29:53Z KAFKA-1690. new java producer needs ssl support as a client. commit 754a121e7582f1452a9ae3a3ab72c58cf284da1d Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T06:02:01Z KAFKA-1690. new java producer needs ssl support as a client. commit 98a90ae9d80ea8f5ab4780569d1c4e301dd16c4e Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T06:18:13Z KAFKA-1690. new java producer needs ssl support as a client. commit 804da7a015be2f98a1bb867ee5d42aa8009a37dd Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T06:31:25Z KAFKA-1690. new java producer needs ssl support as a client. commit ee16e8e6f92ac2baf0e41d3019b7f8aef39b1506 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-11T23:09:01Z KAFKA-1690. new java producer needs ssl support as a client. SSLFactory tests. commit 2dd826be4a6ebe7064cb19ff21fe23950a1bafc2 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-12T23:09:38Z KAFKA-1690. new java producer needs ssl support as a client. Added PrincipalBuilder. commit 2cddad80f6a4a961b6932879448e532dab4e637e Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-15T14:17:37Z KAFKA-1690. new java producer needs ssl support as a client. Addressing reviews. commit ca0456dc01def337ee1711cabd9c4e9df4af61ee Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-20T21:23:29Z KAFKA-1690. new java producer needs ssl support as a client. Addressing reviews. commit 7e3a4cfc58932aab4288677111af52f94c9012b6 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-20T21:37:52Z KAFKA-1690. new java producer needs ssl support as a client. Addressing reviews. commit 9bdceb24f8682184f7fb39578f239a7b6dde Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-21T16:50:52Z KAFKA-1690. new java producer needs ssl support as a client. Fixed minor issues with the patch. commit 65396b5cabeaf61579c6e6422848877fc7a896a9 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-21T17:27:11Z KAFKA-1690. new java producer needs ssl support as a client. Fixed minor issues with the patch. commit b37330a7b4ec3adfba4f0c6e33ab172be03406be Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-05-29T03:57:06Z KAFKA-1690. new java producer needs ssl support as a client. commit fe595fd4fda45ebd7c5da88ee093ab17817bb94d Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-04T01:43:34Z KAFKA-1690. new java producer needs ssl support as a client. commit 247264ce35a04d14b97c87dcb88378ad1dbe0986 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-08T16:07:14Z Merge remote-tracking branch 'refs/remotes/origin/trunk' into KAFKA-1690-V1 commit 050782b9f47f4c61b22ef065ec4798ccbdb962d3 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-16T15:46:05Z KAFKA-1690. Broker side ssl changes. commit 9328ffa464711a835be8935cb09922230e0e1a58 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-20T17:47:01Z KAFKA-1684. SSL for socketServer. commit eda92cb5f9d2ae749903eac5453a6fdb49685964 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T03:01:30Z KAFKA-1690. Added SSLProducerSendTest and fixes to get right port for SSL. commit f10e28b2f2b10d91db9c1aba977fc578b4c4c633 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T03:47:54Z Merge branch 'trunk' into KAFKA-1690-V1 commit f60c95273b3b814792d0da9264a75939049dcc5f Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T04:45:58Z KAFKA-1690. Post merge fixes. commit 8f7ba892502b09cb7cc05d75270352815fb1c42c Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-21T22:35:52Z KAFKA-1690. Added SSLProducerSendTest. commit 0dba29f7bd5489163949641030a98c308f25cb67 Author: Sriharsha Chintalapani har...@hortonworks.com Date: 2015-06-23T16:16:29Z KAFKA-1690. Minor fixes based on patch review comments. commit e44c90e4ce4c36243b1211e25e89d41e82acaf4e Author:
[jira] [Commented] (KAFKA-2411) remove usage of BlockingChannel in the broker
[ https://issues.apache.org/jira/browse/KAFKA-2411?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680061#comment-14680061 ] Ismael Juma commented on KAFKA-2411: [~gwenshap], the time has come for you to pass your knowledge along. :) I would like some feedback on the following commit that updates `KafkaServer`: https://github.com/ijuma/kafka/commit/94d095bf8d6540ada6f733ff05f87cdc358d4ea4 It works correctly as far as I can tell (all tests pass, including `RollingBounceTest`). I am particularly interested in feedback regarding the following (but any and all feedback is welcome): * The various parameters passed to do `Selector` constructor and `Selector.connect` (I reused existing configs, but I am not sure if that's appropriate). * The way I tried to replicate the behaviour of `controllerSocketTimeoutMs`. Regarding `ControllerChannelManager`, we can either keep the existing design with one `RequestSendThread` and `selector` per `toBroker` (in which case the changes are mechanical) or we could have a single thread and `selector` for all brokers. Or something in-between. Thoughts? remove usage of BlockingChannel in the broker - Key: KAFKA-2411 URL: https://issues.apache.org/jira/browse/KAFKA-2411 Project: Kafka Issue Type: Sub-task Components: security Reporter: Jun Rao Assignee: Ismael Juma Fix For: 0.8.3 In KAFKA-1690, we are adding the SSL support at Selector. However, there are still a few places where we use BlockingChannel for inter-broker communication. We need to replace those usage with Selector/NetworkClient to enable inter-broker communication over SSL. Specially, BlockingChannel is currently used in the following places. 1. ControllerChannelManager: for the controller to propagate metadata to the brokers. 2. KafkaServer: for the broker to send controlled shutdown request to the controller. 3. AbstractFetcherThread: for the follower to fetch data from the leader (through SimpleConsumer). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Comment Edited] (KAFKA-2411) remove usage of BlockingChannel in the broker
[ https://issues.apache.org/jira/browse/KAFKA-2411?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680061#comment-14680061 ] Ismael Juma edited comment on KAFKA-2411 at 8/10/15 1:00 PM: - [~gwenshap], the time has come for you to pass your knowledge along. :) I would like some feedback on the following commit that updates `KafkaServer`: https://github.com/ijuma/kafka/commit/f5ef130b7405e5ad5088cdf0fbc633c51860ab47 It works correctly as far as I can tell (all tests pass, including `RollingBounceTest`). I am particularly interested in feedback regarding the following (but any and all feedback is welcome): * The various parameters passed to do `Selector` constructor and `Selector.connect` (I reused existing configs, but I am not sure if that's appropriate). * The way I tried to replicate the behaviour of `controllerSocketTimeoutMs`. Regarding `ControllerChannelManager`, we can either keep the existing design with one `RequestSendThread` and `selector` per `toBroker` (in which case the changes are mechanical) or we could have a single thread and `selector` for all brokers. Or something in-between. Thoughts? was (Author: ijuma): [~gwenshap], the time has come for you to pass your knowledge along. :) I would like some feedback on the following commit that updates `KafkaServer`: https://github.com/ijuma/kafka/commit/94d095bf8d6540ada6f733ff05f87cdc358d4ea4 It works correctly as far as I can tell (all tests pass, including `RollingBounceTest`). I am particularly interested in feedback regarding the following (but any and all feedback is welcome): * The various parameters passed to do `Selector` constructor and `Selector.connect` (I reused existing configs, but I am not sure if that's appropriate). * The way I tried to replicate the behaviour of `controllerSocketTimeoutMs`. Regarding `ControllerChannelManager`, we can either keep the existing design with one `RequestSendThread` and `selector` per `toBroker` (in which case the changes are mechanical) or we could have a single thread and `selector` for all brokers. Or something in-between. Thoughts? remove usage of BlockingChannel in the broker - Key: KAFKA-2411 URL: https://issues.apache.org/jira/browse/KAFKA-2411 Project: Kafka Issue Type: Sub-task Components: security Reporter: Jun Rao Assignee: Ismael Juma Fix For: 0.8.3 In KAFKA-1690, we are adding the SSL support at Selector. However, there are still a few places where we use BlockingChannel for inter-broker communication. We need to replace those usage with Selector/NetworkClient to enable inter-broker communication over SSL. Specially, BlockingChannel is currently used in the following places. 1. ControllerChannelManager: for the controller to propagate metadata to the brokers. 2. KafkaServer: for the broker to send controlled shutdown request to the controller. 3. AbstractFetcherThread: for the follower to fetch data from the leader (through SimpleConsumer). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-1686) Implement SASL/Kerberos
[ https://issues.apache.org/jira/browse/KAFKA-1686?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma updated KAFKA-1686: --- Affects Version/s: (was: 0.9.0) 0.8.2.1 Implement SASL/Kerberos --- Key: KAFKA-1686 URL: https://issues.apache.org/jira/browse/KAFKA-1686 Project: Kafka Issue Type: Sub-task Components: security Affects Versions: 0.8.2.1 Reporter: Jay Kreps Assignee: Sriharsha Chintalapani Fix For: 0.8.3 Implement SASL/Kerberos authentication. To do this we will need to introduce a new SASLRequest and SASLResponse pair to the client protocol. This request and response will each have only a single byte[] field and will be used to handle the SASL challenge/response cycle. Doing this will initialize the SaslServer instance and associate it with the session in a manner similar to KAFKA-1684. When using integrity or encryption mechanisms with SASL we will need to wrap and unwrap bytes as in KAFKA-1684 so the same interface that covers the SSLEngine will need to also cover the SaslServer instance. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-1686) Implement SASL/Kerberos
[ https://issues.apache.org/jira/browse/KAFKA-1686?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma updated KAFKA-1686: --- Fix Version/s: (was: 0.9.0) 0.8.3 Implement SASL/Kerberos --- Key: KAFKA-1686 URL: https://issues.apache.org/jira/browse/KAFKA-1686 Project: Kafka Issue Type: Sub-task Components: security Affects Versions: 0.8.2.1 Reporter: Jay Kreps Assignee: Sriharsha Chintalapani Fix For: 0.8.3 Implement SASL/Kerberos authentication. To do this we will need to introduce a new SASLRequest and SASLResponse pair to the client protocol. This request and response will each have only a single byte[] field and will be used to handle the SASL challenge/response cycle. Doing this will initialize the SaslServer instance and associate it with the session in a manner similar to KAFKA-1684. When using integrity or encryption mechanisms with SASL we will need to wrap and unwrap bytes as in KAFKA-1684 so the same interface that covers the SSLEngine will need to also cover the SaslServer instance. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2210) KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation.
[ https://issues.apache.org/jira/browse/KAFKA-2210?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14679758#comment-14679758 ] Ismael Juma commented on KAFKA-2210: Hi [~parth.brahmbhatt], it would be good to get your feedback on the comments left on the latest round of reviews. If you agree with the suggested changes, but are busy at the moment, I would be happy to incorporate them and submit a revised patch (with you as the author, of course). KafkaAuthorizer: Add all public entities, config changes and changes to KafkaAPI and kafkaServer to allow pluggable authorizer implementation. -- Key: KAFKA-2210 URL: https://issues.apache.org/jira/browse/KAFKA-2210 Project: Kafka Issue Type: Sub-task Components: security Reporter: Parth Brahmbhatt Assignee: Parth Brahmbhatt Fix For: 0.8.3 Attachments: KAFKA-2210.patch, KAFKA-2210_2015-06-03_16:36:11.patch, KAFKA-2210_2015-06-04_16:07:39.patch, KAFKA-2210_2015-07-09_18:00:34.patch, KAFKA-2210_2015-07-14_10:02:19.patch, KAFKA-2210_2015-07-14_14:13:19.patch, KAFKA-2210_2015-07-20_16:42:18.patch, KAFKA-2210_2015-07-21_17:08:21.patch This is the first subtask for Kafka-1688. As Part of this jira we intend to agree on all the public entities, configs and changes to existing kafka classes to allow pluggable authorizer implementation. Please see KIP-11 https://cwiki.apache.org/confluence/display/KAFKA/KIP-11+-+Authorization+Interface for detailed design. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-2211) KafkaAuthorizer: Add simpleACLAuthorizer implementation.
[ https://issues.apache.org/jira/browse/KAFKA-2211?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma updated KAFKA-2211: --- Fix Version/s: 0.8.3 KafkaAuthorizer: Add simpleACLAuthorizer implementation. Key: KAFKA-2211 URL: https://issues.apache.org/jira/browse/KAFKA-2211 Project: Kafka Issue Type: Sub-task Components: security Reporter: Parth Brahmbhatt Assignee: Parth Brahmbhatt Fix For: 0.8.3 Attachments: KAFKA-2211.patch Subtask-2 for Kafka-1688. Please see KIP-11 to get details on out of box SimpleACLAuthorizer implementation https://cwiki.apache.org/confluence/display/KAFKA/KIP-11+-+Authorization+Interface. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-1686) Implement SASL/Kerberos
[ https://issues.apache.org/jira/browse/KAFKA-1686?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14679764#comment-14679764 ] Ismael Juma commented on KAFKA-1686: [~sriharsha], I changed the target version to 0.8.3. Please let me know if you disagree with this. Implement SASL/Kerberos --- Key: KAFKA-1686 URL: https://issues.apache.org/jira/browse/KAFKA-1686 Project: Kafka Issue Type: Sub-task Components: security Affects Versions: 0.8.2.1 Reporter: Jay Kreps Assignee: Sriharsha Chintalapani Fix For: 0.8.3 Implement SASL/Kerberos authentication. To do this we will need to introduce a new SASLRequest and SASLResponse pair to the client protocol. This request and response will each have only a single byte[] field and will be used to handle the SASL challenge/response cycle. Doing this will initialize the SaslServer instance and associate it with the session in a manner similar to KAFKA-1684. When using integrity or encryption mechanisms with SASL we will need to wrap and unwrap bytes as in KAFKA-1684 so the same interface that covers the SSLEngine will need to also cover the SaslServer instance. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-1683) Implement a session concept in the socket server
[ https://issues.apache.org/jira/browse/KAFKA-1683?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma updated KAFKA-1683: --- Fix Version/s: 0.8.3 Implement a session concept in the socket server -- Key: KAFKA-1683 URL: https://issues.apache.org/jira/browse/KAFKA-1683 Project: Kafka Issue Type: Sub-task Components: security Affects Versions: 0.9.0 Reporter: Jay Kreps Assignee: Gwen Shapira Fix For: 0.8.3 Attachments: KAFKA-1683.patch, KAFKA-1683.patch To implement authentication we need a way to keep track of some things between requests. The initial use for this would be remembering the authenticated user/principle info, but likely more uses would come up (for example we will also need to remember whether and which encryption or integrity measures are in place on the socket so we can wrap and unwrap writes and reads). I was thinking we could just add a Session object that might have a user field. The session object would need to get added to RequestChannel.Request so it is passed down to the API layer with each request. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
Re: [jira] [Commented] (KAFKA-1690) new java producer needs ssl support as a client
I was running a Kafka cluster with the latest SSL patch over the weekend with IBM JRE, and it has been running fine without any issues. There was light load on the cluster throughout and intermittent heavy load, all using SSL clients. However I am seeing an intermittent unit test hang in org.apache.kafka.common.network.SSLSelectorTest.testRenegotiate(). I am not sure if that is related to the IBM JRE I am using for the build. It looks like data left in appReadBuffer after a handshake may not get processed if no more data arrives on the network, causing the test to loop forever. It will be good if this can be fixed (or at least the test commented out) before the code is committed to avoid breaking the build. Even though there are quite a few outstanding review comments, I do agree that being such a big patch, it will be good to commit the code soon and work on minor issues afterwards. On Mon, Aug 10, 2015 at 12:19 AM, Jun Rao (JIRA) j...@apache.org wrote: [ https://issues.apache.org/jira/browse/KAFKA-1690?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14679397#comment-14679397 ] Jun Rao commented on KAFKA-1690: [~rsivaram], this jira is getting pretty close to be committed. Could you test this out on an IBM jvm to see if there is any issue especially with respect to the usage of the sendfile api? new java producer needs ssl support as a client --- Key: KAFKA-1690 URL: https://issues.apache.org/jira/browse/KAFKA-1690 Project: Kafka Issue Type: Sub-task Reporter: Joe Stein Assignee: Sriharsha Chintalapani Fix For: 0.8.3 Attachments: KAFKA-1690.patch, KAFKA-1690.patch, KAFKA-1690_2015-05-10_23:20:30.patch, KAFKA-1690_2015-05-10_23:31:42.patch, KAFKA-1690_2015-05-11_16:09:36.patch, KAFKA-1690_2015-05-12_16:20:08.patch, KAFKA-1690_2015-05-15_07:18:21.patch, KAFKA-1690_2015-05-20_14:54:35.patch, KAFKA-1690_2015-05-21_10:37:08.patch, KAFKA-1690_2015-06-03_18:52:29.patch, KAFKA-1690_2015-06-23_13:18:20.patch, KAFKA-1690_2015-07-20_06:10:42.patch, KAFKA-1690_2015-07-20_11:59:57.patch, KAFKA-1690_2015-07-25_12:10:55.patch -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-2412) Documentation bug: Add information for key.serializer and value.serializer to New Producer Config sections
[ https://issues.apache.org/jira/browse/KAFKA-2412?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14679886#comment-14679886 ] darshan kumar commented on KAFKA-2412: -- Hi, I have cloned the repository, git clone https://git-wip-us.apache.org/repos/asf/kafka.git kafka. I will find the options in javadoc and producer.java examples for key.serializer and value.serializer. Thank you. Documentation bug: Add information for key.serializer and value.serializer to New Producer Config sections -- Key: KAFKA-2412 URL: https://issues.apache.org/jira/browse/KAFKA-2412 Project: Kafka Issue Type: Bug Reporter: Jeremy Fields Priority: Minor Labels: newbie As key.serializer and value.serializer are required options when using the new producer, they should be mentioned in the documentation ( here and svn http://kafka.apache.org/documentation.html#newproducerconfigs ) Appropriate values for these options exist in javadoc and producer.java examples; however, not everyone is reading those, as is the case for anyone setting up a producer.config file for mirrormaker. A sensible default should be suggested, such as org.apache.kafka.common.serialization.StringSerializer Or at least a mention of the key.serializer and value.serializer options along with a link to javadoc Thanks -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Commented] (KAFKA-1695) Authenticate connection to Zookeeper
[ https://issues.apache.org/jira/browse/KAFKA-1695?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanelfocusedCommentId=14680028#comment-14680028 ] Ismael Juma commented on KAFKA-1695: [~parth.brahmbhatt], do you know when the new release for `zkClient` wil be out? Do I understand correctly that it's a blocker for this work? Trying to figure out if we should be targetting this for 0.8.3. Authenticate connection to Zookeeper Key: KAFKA-1695 URL: https://issues.apache.org/jira/browse/KAFKA-1695 Project: Kafka Issue Type: Sub-task Components: security Reporter: Jay Kreps Assignee: Parth Brahmbhatt We need to make it possible to secure the Zookeeper cluster Kafka is using. This would make use of the normal authentication ZooKeeper provides. ZooKeeper supports a variety of authentication mechanisms so we will need to figure out what has to be passed in to the zookeeper client. The intention is that when the current round of client work is done it should be possible to run without clients needing access to Zookeeper so all we need here is to make it so that only the Kafka cluster is able to read and write to the Kafka znodes (we shouldn't need to set any kind of acl on a per-znode basis). -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-1685) Implement TLS/SSL tests
[ https://issues.apache.org/jira/browse/KAFKA-1685?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma updated KAFKA-1685: --- Fix Version/s: (was: 0.9.0) 0.8.3 Implement TLS/SSL tests --- Key: KAFKA-1685 URL: https://issues.apache.org/jira/browse/KAFKA-1685 Project: Kafka Issue Type: Sub-task Components: security Affects Versions: 0.8.2.1 Reporter: Jay Kreps Assignee: Sriharsha Chintalapani Fix For: 0.8.3 We need to write a suite of unit tests for TLS authentication. This should be doable with a junit integration test. We can use the simple authorization plugin with only a single user whitelisted. The test can start the server and then connects with and without TLS and validates that access is only possible when authenticated. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
[jira] [Updated] (KAFKA-1685) Implement TLS/SSL tests
[ https://issues.apache.org/jira/browse/KAFKA-1685?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Ismael Juma updated KAFKA-1685: --- Affects Version/s: (was: 0.9.0) 0.8.2.1 Implement TLS/SSL tests --- Key: KAFKA-1685 URL: https://issues.apache.org/jira/browse/KAFKA-1685 Project: Kafka Issue Type: Sub-task Components: security Affects Versions: 0.8.2.1 Reporter: Jay Kreps Assignee: Sriharsha Chintalapani Fix For: 0.8.3 We need to write a suite of unit tests for TLS authentication. This should be doable with a junit integration test. We can use the simple authorization plugin with only a single user whitelisted. The test can start the server and then connects with and without TLS and validates that access is only possible when authenticated. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
