kafka-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Jun Rao" <jun...@gmail.com>
Subject Re: Review Request 14730: Patch for KAFKA-1001
Date Tue, 22 Oct 2013 17:52:22 GMT

-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/14730/#review27309
-----------------------------------------------------------



core/src/main/scala/kafka/server/AbstractFetcherManager.scala
<https://reviews.apache.org/r/14730/#comment53135>

    This method can be written in a clearer way. There is code there that does the equivalent
of collection.groupBy. If we change the input a set, we can first make a pass to compute the
fetcherId and then use collection.groupBy to group the set by BrokerFetcherId. Finally, we
can add the set per BokerFetcherId to the fetcherThreadMap and call addPartitions.



core/src/main/scala/kafka/server/AbstractFetcherManager.scala
<https://reviews.apache.org/r/14730/#comment53136>

    This needs to be protected by the mapLock. For simplicity, we can probably just hold the
lock in the whole logic. Adding all partitions should be cheap and there is no strong reason
to optimize that.



core/src/main/scala/kafka/server/AbstractFetcherManager.scala
<https://reviews.apache.org/r/14730/#comment53138>

    This can be written as 
    fetcherThreadMap(brokerAndFetcherId).



core/src/main/scala/kafka/server/AbstractFetcherThread.scala
<https://reviews.apache.org/r/14730/#comment53141>

    I recommend that we hold the lock for the whole method. This may not be necessary for
the logic we have now. However, this may change if the logic evolves in the future. Also,
iterating an in-memory data structure should be cheap. So, optimizing the locking period is
not necessary.



core/src/main/scala/kafka/server/ReplicaManager.scala
<https://reviews.apache.org/r/14730/#comment53142>

    Hmm, I don't see the code for marking them as stopped?



core/src/main/scala/kafka/server/ReplicaManager.scala
<https://reviews.apache.org/r/14730/#comment53145>

    I think we need to move replicaStateChangeLock to here. Otherwise, the checking and the
updating of controllerEpoch may not be atomic.



core/src/main/scala/kafka/server/ReplicaManager.scala
<https://reviews.apache.org/r/14730/#comment53144>

    Could we use a better val name? Something like leaderPartitionInfos?



core/src/main/scala/kafka/server/ReplicaManager.scala
<https://reviews.apache.org/r/14730/#comment53143>

    Could we use named fields instead of _2?



core/src/main/scala/kafka/server/ReplicaManager.scala
<https://reviews.apache.org/r/14730/#comment53148>

    Can this logic be moved to becomeLeaderOrFollower since it's duplicated in becomeFollower()?



core/src/main/scala/kafka/server/ReplicaManager.scala
<https://reviews.apache.org/r/14730/#comment53146>

    Since we serialize the processing of all requests from the controller, we should never
hit the code here. So, we should change the logging to WARN and change the logging message
accordingly.



core/src/main/scala/kafka/server/ReplicaManager.scala
<https://reviews.apache.org/r/14730/#comment53149>

    Are we expecting any exceptions here? If so, what about the error code in the response?


- Jun Rao


On Oct. 21, 2013, 8:35 p.m., Guozhang Wang wrote:
> 
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/14730/
> -----------------------------------------------------------
> 
> (Updated Oct. 21, 2013, 8:35 p.m.)
> 
> 
> Review request for kafka.
> 
> 
> Bugs: KAFKA-1001
>     https://issues.apache.org/jira/browse/KAFKA-1001
> 
> 
> Repository: kafka
> 
> 
> Description
> -------
> 
> Incorporated review comments
> 
> 
> Diffs
> -----
> 
>   core/src/main/scala/kafka/cluster/Partition.scala 5ccecd179d33abfc14dcefc35dd68de7474c6978

>   core/src/main/scala/kafka/common/ErrorMapping.scala 153bc0b078d21200c02c47dd5ad9b7a7e3326ec4

>   core/src/main/scala/kafka/consumer/ConsumerFetcherManager.scala 566ca46d113ee7da4b38ee57302ba183b59ab5d6

>   core/src/main/scala/kafka/consumer/ConsumerFetcherThread.scala dda0a8f041f242bf8a501a8cbd2b9c0258323f96

>   core/src/main/scala/kafka/log/LogManager.scala 47197153c5d3797d2e2a2f9539d9cd55501468e3

>   core/src/main/scala/kafka/server/AbstractFetcherManager.scala 15b7bd31446ffb97b8ed0fa6461649a01d81c7e9

>   core/src/main/scala/kafka/server/AbstractFetcherThread.scala c64260f12bdd6b6c964875e1f3873156442e44e1

>   core/src/main/scala/kafka/server/ReplicaManager.scala ee1cc0cf451b691eb91d9158ca765aeb60fc3dc8

> 
> Diff: https://reviews.apache.org/r/14730/diff/
> 
> 
> Testing
> -------
> 
> 
> Thanks,
> 
> Guozhang Wang
> 
>


Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message