-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/28769/#review66944
-----------------------------------------------------------
Thanks for the patch. It's a lot work! A few general comments below, in
addition to the more detailed comments.
1. Formatting: add a space after comma in function signature and function
calls. So instead of
def foo(a: A,b: B)
use
def foo(a: A, b: B)
2. In some of the files, imports can be optimized.
3. Some new files are missing the license header.
4. It seems that the client needs to know the security protocol in addition to
the port. Otherwise, it doesn't know which protocol to use to establish the
socket connection. So, perhaps the broker.list should now be
protocol://host:port?
core/src/main/scala/kafka/client/ClientUtils.scala
<https://reviews.apache.org/r/28769/#comment110759>
Is protocolType used?
core/src/main/scala/kafka/cluster/Broker.scala
<https://reviews.apache.org/r/28769/#comment110770>
It would be good to document the format for both v1 and v2.
Also, would it be cleaner if in v2, we remove the host/port fields and only
include the endpoints field? Currently, v2 writes host/port, but doesn't use it
itself. So, it's really just for backward compatibility. Since we have the
"use.new.wire.protocol" config, we can let each broker register in ZK using v2
format only after that config is set to true (when all brokers are upgraded to
the new binary).
core/src/main/scala/kafka/cluster/Broker.scala
<https://reviews.apache.org/r/28769/#comment110676>
Is this used?
core/src/main/scala/kafka/cluster/Broker.scala
<https://reviews.apache.org/r/28769/#comment110689>
Instead of converting a seq to a map, perhaps it's simpler to just do
seq.find(predicate).
core/src/main/scala/kafka/cluster/EndPoint.scala
<https://reviews.apache.org/r/28769/#comment110771>
Do we need the dot after -?
core/src/main/scala/kafka/cluster/EndPoint.scala
<https://reviews.apache.org/r/28769/#comment110772>
Do we want to support IPV6 format of host?
core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala
<https://reviews.apache.org/r/28769/#comment110774>
securityProtocol doesn't seem to be used here.
core/src/main/scala/kafka/network/BlockingChannel.scala
<https://reviews.apache.org/r/28769/#comment110776>
Adding this as an error log may pollute the log if a broker is unreachable
and a client keeps trying to connect to it in a certain window (e.g., a replica
fetcher keeps fetching from the old leader until the new leader info is
obtained from the controller).
core/src/main/scala/kafka/network/SocketServer.scala
<https://reviews.apache.org/r/28769/#comment110680>
Should this be private?
core/src/main/scala/kafka/network/SocketServer.scala
<https://reviews.apache.org/r/28769/#comment110678>
I think each acceptor needs its own set of processors. We probably need to
change the doc for num.network.threads to indicate that it's per endpoint.
core/src/main/scala/kafka/producer/ProducerPool.scala
<https://reviews.apache.org/r/28769/#comment110777>
Do we need to recreate the endpoint here?
core/src/main/scala/kafka/server/KafkaConfig.scala
<https://reviews.apache.org/r/28769/#comment110687>
To make the upgrade easier, it may be useful to support the old configs in
a backward compatible way. For example, if the user doesn't specify listeners,
but specifies host.name and port, we can automatically set "listeners" to a
single listener based on host.name and port. Ditto for advertised listeners.
Also, it would be useful to verify that all ports in the listeners are
unique.
core/src/main/scala/kafka/server/KafkaConfig.scala
<https://reviews.apache.org/r/28769/#comment110717>
In KafkaApi, it seems that we assume there is at most one endpoint per
protocol (since we look for endpoint by protocol). If so, it would be good to
validate that at config time.
core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
<https://reviews.apache.org/r/28769/#comment110779>
Is this change necessary?
- Jun Rao
On Jan. 6, 2015, 7:46 p.m., Gwen Shapira wrote:
>
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/28769/
> -----------------------------------------------------------
>
> (Updated Jan. 6, 2015, 7:46 p.m.)
>
>
> Review request for kafka.
>
>
> Bugs: KAFKA-1809
> https://issues.apache.org/jira/browse/KAFKA-1809
>
>
> Repository: kafka
>
>
> Description
> -------
>
> first commit of refactoring.
>
>
> changed topicmetadata to include brokerendpoints and fixed few unit tests
>
>
> fixing systest and support for binding to default address
>
>
> fixed system tests
>
>
> fix default address binding and ipv6 support
>
>
> fix some issues regarding endpoint parsing. Also, larger segments for systest
> make the validation much faster
>
>
> added link to security wiki in doc
>
>
> fixing unit test after rename of ProtocolType to SecurityProtocol
>
>
> Following Joe's advice, added security protocol enum on client side, and
> modified protocol to use ID instead of string.
>
>
> validate producer config against enum
>
>
> add a second protocol for testing and modify SocketServerTests to check on
> multi-ports
>
>
> Reverted the metadata request changes and removed the explicit security
> protocol argument. Instead the socketserver will determine the protocol based
> on the port and add this to the request
>
>
> bump version for UpdateMetadataRequest and added support for rolling upgrades
> with new config
>
>
> following tests - fixed LeaderAndISR protocol and ZK registration for
> backward compatibility
>
>
> cleaned up some changes that were actually not necessary. hopefully making
> this patch slightly easier to review
>
>
> undoing some changes that don't belong here
>
>
> bring back config lost in cleanup
>
>
> fixes neccessary for an all non-plaintext cluster to work
>
>
> Diffs
> -----
>
> clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
> 525b95e98010cd2053eacd8c321d079bcac2f910
>
> clients/src/main/java/org/apache/kafka/common/protocol/SecurityProtocol.java
> PRE-CREATION
> clients/src/main/java/org/apache/kafka/common/utils/Utils.java
> 527dd0f9c47fce7310b7a37a9b95bf87f1b9c292
> clients/src/test/java/org/apache/kafka/common/utils/ClientUtilsTest.java
> 6e37ea553f73d9c584641c48c56dbf6e62ba5f88
> clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
> a39fab532f73148316a56c0f8e9197f38ea66f79
> config/server.properties b0e4496a8ca736b6abe965a430e8ce87b0e8287f
> core/src/main/scala/kafka/admin/AdminUtils.scala
> 28b12c7b89a56c113b665fbde1b95f873f8624a3
> core/src/main/scala/kafka/admin/TopicCommand.scala
> 285c0333ff43543d3e46444c1cd9374bb883bb59
> core/src/main/scala/kafka/api/ConsumerMetadataResponse.scala
> 84f60178f6ebae735c8aa3e79ed93fe21ac4aea7
> core/src/main/scala/kafka/api/LeaderAndIsrRequest.scala
> 4ff7e8f8cc695551dd5d2fe65c74f6b6c571e340
> core/src/main/scala/kafka/api/TopicMetadata.scala
> 0190076df0adf906ecd332284f222ff974b315fc
> core/src/main/scala/kafka/api/TopicMetadataResponse.scala
> 92ac4e687be22e4800199c0666bfac5e0059e5bb
> core/src/main/scala/kafka/api/UpdateMetadataRequest.scala
> 530982e36b17934b8cc5fb668075a5342e142c59
> core/src/main/scala/kafka/client/ClientUtils.scala
> ebba87f0566684c796c26cb76c64b4640a5ccfde
> core/src/main/scala/kafka/cluster/Broker.scala
> 0060add008bb3bc4b0092f2173c469fce0120be6
> core/src/main/scala/kafka/cluster/BrokerEndPoint.scala PRE-CREATION
> core/src/main/scala/kafka/cluster/EndPoint.scala PRE-CREATION
> core/src/main/scala/kafka/cluster/SecurityProtocol.scala PRE-CREATION
> core/src/main/scala/kafka/common/BrokerEndPointNotAvailableException.scala
> PRE-CREATION
> core/src/main/scala/kafka/consumer/ConsumerConfig.scala
> 9ebbee6c16dc83767297c729d2d74ebbd063a993
> core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala
> b9e2bea7b442a19bcebd1b350d39541a8c9dd068
> core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala
> ee6139c901082358382c5ef892281386bf6fc91b
> core/src/main/scala/kafka/consumer/ZookeeperConsumerConnector.scala
> 191a8677444e53b043e9ad6e94c5a9191c32599e
> core/src/main/scala/kafka/controller/ControllerChannelManager.scala
> eb492f00449744bc8d63f55b393e2a1659d38454
> core/src/main/scala/kafka/controller/KafkaController.scala
> 66df6d2fbdbdd556da6bea0df84f93e0472c8fbf
> core/src/main/scala/kafka/javaapi/ConsumerMetadataResponse.scala
> 1b28861cdf7dfb30fc696b54f8f8f05f730f4ece
> core/src/main/scala/kafka/javaapi/TopicMetadata.scala
> f384e04678df10a5b46a439f475c63371bf8e32b
> core/src/main/scala/kafka/javaapi/TopicMetadataRequest.scala
> b0b7be14d494ae8c87f4443b52db69d273c20316
> core/src/main/scala/kafka/network/BlockingChannel.scala
> 6e2a38eee8e568f9032f95c75fa5899e9715b433
> core/src/main/scala/kafka/network/RequestChannel.scala
> 7b1db3dbbb2c0676f166890f566c14aa248467ab
> core/src/main/scala/kafka/network/SocketServer.scala
> e451592fe358158548117f47a80e807007dd8b98
> core/src/main/scala/kafka/producer/ProducerPool.scala
> 43df70bb461dd3e385e6b20396adef3c4016a3fc
> core/src/main/scala/kafka/server/AbstractFetcherManager.scala
> 20c00cb8cc2351950edbc8cb1752905a0c26e79f
> core/src/main/scala/kafka/server/AbstractFetcherThread.scala
> 8c281d4668f92eff95a4a5df3c03c4b5b20e7095
> core/src/main/scala/kafka/server/KafkaApis.scala
> 2a1c0326b6e6966d8b8254bd6a1cb83ad98a3b80
> core/src/main/scala/kafka/server/KafkaConfig.scala
> 6e26c5436feb4629d17f199011f3ebb674aa767f
> core/src/main/scala/kafka/server/KafkaHealthcheck.scala
> 4acdd70fe9c1ee78d6510741006c2ece65450671
> core/src/main/scala/kafka/server/KafkaServer.scala
> 1bf7d10cef23a77e716666eb16bf6d0e68bc4ebe
> core/src/main/scala/kafka/server/MetadataCache.scala
> bf81a1ab88c14be8697b441eedbeb28fa0112643
> core/src/main/scala/kafka/server/ReplicaFetcherManager.scala
> 351dbbad3bdb709937943b336a5b0a9e0162a5e2
> core/src/main/scala/kafka/server/ReplicaFetcherThread.scala
> 6879e730282185bda3d6bc3659cb15af0672cecf
> core/src/main/scala/kafka/server/ReplicaManager.scala
> e58fbb922e93b0c31dff04f187fcadb4ece986d7
> core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala
> d1e7c434e77859d746b8dc68dd5d5a3740425e79
> core/src/main/scala/kafka/tools/ReplicaVerificationTool.scala
> ba6ddd7a909df79a0f7d45e8b4a2af94ea0fceb6
> core/src/main/scala/kafka/tools/SimpleConsumerShell.scala
> b4f903b6c7c3bb725cac7c05eb1f885906413c4d
> core/src/main/scala/kafka/tools/UpdateOffsetsInZK.scala
> 111c9a8b94ce45d95551482e9fd3f8c1cccbf548
> core/src/main/scala/kafka/utils/Utils.scala
> 738c1af9ef5de16fdf5130daab69757a14c48b5c
> core/src/main/scala/kafka/utils/ZkUtils.scala
> 56e3e88e0cc6d917b0ffd1254e173295c1c4aabd
> core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
> 5ec613cdb50b93bfe4477e89554dfc3768759b18
> core/src/test/scala/integration/kafka/api/ProducerSendTest.scala
> 6196060edf9f1650720ec916f88933953a1daa2c
> core/src/test/scala/other/kafka/TestOffsetManager.scala
> 41f334d48897b3027ed54c58bbf4811487d3b191
> core/src/test/scala/unit/kafka/KafkaConfigTest.scala
> 4d36b8b1173f60d43463c13c9d8c1275a84c8c28
> core/src/test/scala/unit/kafka/admin/AddPartitionsTest.scala
> 1bf2667f47853585bc33ffb3e28256ec5f24ae84
> core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala
> cd16ced5465d098be7a60498326b2a98c248f343
> core/src/test/scala/unit/kafka/cluster/BrokerTest.scala PRE-CREATION
> core/src/test/scala/unit/kafka/consumer/ConsumerIteratorTest.scala
> c0355cc0135c6af2e346b4715659353a31723b86
> core/src/test/scala/unit/kafka/integration/FetcherTest.scala
> 25845abbcad2e79f56f729e59239b738d3ddbc9d
> core/src/test/scala/unit/kafka/integration/TopicMetadataTest.scala
> 35dc071b1056e775326981573c9618d8046e601d
> core/src/test/scala/unit/kafka/network/SocketServerTest.scala
> 5f4d85254c384dcc27a5a84f0836ea225d3a901a
> core/src/test/scala/unit/kafka/producer/AsyncProducerTest.scala
> 1db6ac329f7b54e600802c8a623f80d159d4e69b
> core/src/test/scala/unit/kafka/producer/SyncProducerTest.scala
> d60d8e0f49443f4dc8bc2cad6e2f951eda28f5cb
> core/src/test/scala/unit/kafka/server/AdvertiseBrokerTest.scala
> f0c4a56b61b4f081cf4bee799c6e9c523ff45e19
> core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
> 2377abe4933e065d037828a214c3a87e1773a8ef
> core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala
> c2ba07c5fdbaf0e65ca033b2e4d88f45a8a15b2e
> core/src/test/scala/unit/kafka/server/LogOffsetTest.scala
> c06ee756bf0fe07e5d3c92823a476c960b37afd6
> core/src/test/scala/unit/kafka/utils/TestUtils.scala
> 94d0028d8c4907e747aa8a74a13d719b974c97bf
> system_test/replication_testsuite/testcase_1/testcase_1_properties.json
> 0c6d7a316cc6b51ac0755ca03558507db0706c31
> system_test/utils/kafka_system_test_utils.py
> 41d511cbc310fa87e0f2cd2f772e479e8e3ae4e2
>
> Diff: https://reviews.apache.org/r/28769/diff/
>
>
> Testing
> -------
>
>
> Thanks,
>
> Gwen Shapira
>
>
