[ https://issues.apache.org/jira/browse/KAFKA-2045?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14385500#comment-14385500 ]
Rajiv Kurian edited comment on KAFKA-2045 at 3/28/15 7:40 PM: -------------------------------------------------------------- [~jkreps] Totally agree with you on the concerns with a re-write. I am sure I'll end up re-using most of the code, otherwise it will take too long in any case. But given this is just a prototype, I want the freedom to be able to make changes without being bound by the existing architecture and class hierarchy of the client. Even if I do re-implement some of the parts I'll make sure that the client can (a) Do metadata requests so it can react to leaders moving etc. (b) Actually read from multiple topic/partitions spread across multiple brokers and not just a single broker. Again since this is just a rewrite with the sole purpose of exploring possible performance improvements there can be mainly two consequences: i) It shows no improvements: In that case we end up not spending too much time changing the current code, and the hacky code just gets us to this conclusion faster. ii) It shows interesting improvements: If this were true, we can afford to spend some time seeing which things actually improved performance and make a call on how to integrate best. It might be counterproductive to look at the current client implementation and look at the % of time spent in each of the bottlenecks because those numbers are a consequence of the current memory layout. For example if we do an on the fly CRC check and decompression - CRC check time might go up a bit because now we are not striding over a contiguous ByteBuffer in one sweep. Right now the current client has this pattern --- CRC check on Message1 --> CRC check on Message2 .... --> CRC check on MessageN --> Hand message 1 to consumer .... --> Hand message N to consumer. Instead with the current proposal we will have a pattern of ----- Do CRC on a Message1 --> Hand Message1 to consumer --> Do CRC on a Message2 --> Hand Message2 to the consumer . So the CRC checks are separated by potential (certain?) cache floundering during the handling of the message by the consumer. On the other hand from the perspective of the consumer, the pattern looks like this -- Do CRC and validation on all messages starting with 1 to N --> Hand messages 1 to N to client. Now by the time the Kafka consumer is done with validating and deserializing message N, message 1 is possibly already out of the cache. With the new approach since we hand over a message right after validating it, we give the consumer a hot in cache message, which might improve the consumer processing enough to offset for the loss in CRC striding efficiency. Or it may not. It might just turn out that doing the CRC validation upfront is just a pure win since all the CRC tables will be in cache etc and striding access for the CRC math is worth an extra iteration of the ByteBuffer contents. But it is might still be more profitable to elide copies and prevent creation of objects by doing on the fly decoding and handing out indexes into the actual response ByteBuffer. This result might further be affected by how expensive the deserialization and processing of the message is. If the message is a bloated JSON encoded object that is de-serialized into a POJO and then processed really slowly then none of this will probably matter. On the other hand if the message is a compact and binary encoded and can be processed with minimal cache misses, this stuff might add up. My point is that basing the TODOs on the current profile may not be optimal because the profile is a massive consequence of the current layout and allocation patterns. Also the profile will give %s and we might be able to keep the same %s but just still reduce the overall time taken for the entire consumer processing cycle. Just to belabor the point even further, the current hash map implementations might suffer so many cache misses that they mask an underlying improvement opportunity for the data in the maps. Switching to compact primitive arrays based open hash maps might surface that opportunity again. Is there a performance test that is used to keep track of the new Consumer's performance? If so maybe I can wrap that in a JMH suite and re-use that to test improvements? was (Author: rzidane): [~jkreps] Totally agree with you on the concerns with a re-write. I am sure I'll end up re-using most of the code, otherwise it will take too long in any case. But given this is just a prototype, I want the freedom to be able to make changes without being bound by the existing architecture and class hierarchy of the client. Even if I do re-implement some of the parts I'll make sure that the client can (a) Do metadata requests so it can react to leaders moving etc. (b) Actually read from multiple topic/partitions spread across multiple brokers and not just a single broker. Again since this is just a rewrite with the sole purpose of exploring possible performance improvements there can be mainly two consequences: i) It shows no improvements: In that case we end up not spending too much time changing the current code, and the hacky code just gets us to this conclusion faster. ii) It shows interesting improvements: If this were true, we can afford to spend some time seeing which things actually improved performance and make a call on how to integrate best. It might be counterproductive to look at the current client implementation and look at the % of time spent in each of the bottlenecks because those numbers are a consequence of the current memory layout. For example if we do an on the fly CRC check and decompression - CRC check time might go up a bit because now we are not striding over a contiguous ByteBuffer in one sweep. Right now the current client has this pattern --- CRC check on Message1 --> CRC check on Message2 .... --> CRC check on MessageN --> Hand message 1 to consumer .... --> Hand message N to consumer. Instead with the current proposal we will have a pattern of ----- Do CRC on a Message1 --> Hand Message1 to consumer --> Do CRC on a Message2 --> Hand Message2 to the consumer . So the CRC checks are separated by potential (certain?) cache floundering during the handling of the message by the client. On the other hand from the perspective of the consumer, the pattern looks like this -- Do CRC and validation on all messages starting with 1 to N --> Hand messages 1 to N to client. Now by the time the Kafka consumer is done with validating and deserializing message N, message 1 is possibly already out of the cache. With the new approach since we hand over a message right after validating it, we give the consumer a hot in cache message, which might improve the consumer processing enough to offset for the loss in CRC striding efficiency. Or it may not. It might just turn out that doing the CRC validation upfront is just a pure win since all the CRC tables will be in cache etc and striding access for the CRC math is worth an extra iteration of the ByteBuffer contents. But it is might still be more profitable to elide copies and prevent creation of objects by doing on the fly decoding and handing out indexes into the actual response ByteBuffer. This result might further be affected by how expensive the deserialization and processing of the message is. If the message is a bloated JSON encoded object that is de-serialized into a POJO and then processed really slowly then none of this will probably matter. On the other hand if the message is a compact and binary encoded and can be processed with minimal cache misses, this stuff might add up. My point is that basing the TODOs on the current profile may not be optimal because the profile is a massive consequence of the current layout and allocation patterns. Also the profile will give %s and we might be able to keep the same %s but just still reduce the overall time taken for the entire consumer processing cycle. Just to belabor the point even further, the current hash map implementations might suffer so many cache misses that they mask an underlying improvement opportunity for the data in the maps. Switching to compact primitive arrays based open hash maps might surface that opportunity again. Is there a performance test that is used to keep track of the new Consumer's performance? If so maybe I can wrap that in a JMH suite and re-use that to test improvements? > Memory Management on the consumer > --------------------------------- > > Key: KAFKA-2045 > URL: https://issues.apache.org/jira/browse/KAFKA-2045 > Project: Kafka > Issue Type: Sub-task > Reporter: Guozhang Wang > > We need to add the memory management on the new consumer like we did in the > new producer. This would probably include: > 1. byte buffer re-usage for fetch response partition data. > 2. byte buffer re-usage for on-the-fly de-compression. -- This message was sent by Atlassian JIRA (v6.3.4#6332)
