Hey Jun, Thanks for your comment! Please see my reply below.
On Wed, Mar 15, 2017 at 9:45 PM, Jun Rao <j...@confluent.io> wrote: > Hi, Dong, > > Thanks for the reply. > > 10. Could you comment on that? > Sorry, I missed that comment. Good point. I think the log segments in topicPartition.move directory will be subject to log truncation, log retention and log cleaning in the same way as the log segments in the source log directory. I just specified this inthe KIP. > > 11.2 "I am concerned that the ChangeReplicaDirRequest would be lost if > broker > restarts after it sends ChangeReplicaDirResponse but before it receives > LeaderAndIsrRequest." > > In that case, the reassignment tool could detect that through > DescribeDirsRequest > and issue ChangeReplicaDirRequest again, right? In the common case, this is > probably not needed and we only need to write each replica once. > > My main concern with the approach in the current KIP is that once a new > replica is created in the wrong log dir, the cross log directory movement > may not catch up until the new replica is fully bootstrapped. So, we end up > writing the data for the same replica twice. > I agree with your concern. My main concern is that it is a bit weird if ChangeReplicaDirResponse can not guarantee success and the tool needs to rely on DescribeDirResponse to see if it needs to send ChangeReplicaDirRequest again. How about this: If broker doesn't not have already replica created for the specified topicParition when it receives ChangeReplicaDirRequest, it will reply ReplicaNotAvailableException AND remember (replica, destination log directory) pair in memory to create the replica in the specified log directory. > > 11.3 Are you saying the value in --throttle will be used to set both > intra.broker.throttled.rate and leader.follower.replication. > throttled.replicas? > No. --throttle will be used to only to set leader.follower.replication as it does now. I think we do not need any option in the kafka-reassignment-partitions.sh to specify intra.broker.throttled.rate. User canset it in broker config or dynamically using kafka-config.sh. Does this sound OK? > > 12.2 If the user only wants to check one topic, the tool could do the > filtering on the client side, right? My concern with having both log_dirs > and topics is the semantic. For example, if both are not empty, do we > return the intersection or the union? > Yes the tool could filter on the client side. But the purpose of having this field is to reduce response side in case broker has a lot of topics. The both fields are used as filter and the result is intersection. Do you think this semantic is confusing or counter-intuitive? > > 12.3. Yes, firstOffset may not be useful. > > 14. Hmm, I would think moving data across log dirs will be io bound. We > also have num.recovery.threads.per.data.dir, which defaults to 1. So, > having num.replica.move.threads defaults to # log dirs or half of that (to > account for ios on both source and target) seems reasonable. Is a magical > value of 3 going to be better? Does that work with only 2 log dirs? There > will always be cases when the user needs to customize the value. We just > need a reasonable default to cover the common case. > If the throughput of moving data across dir doesn't not increase with number of threads, I think we should provide config num.replica.move.thread.per.log.dir and give it default value of 1. That works in the same way as num.recovery.threads.per.data.dir. But I think the replica movement is not necessarily IO bound if broker is using SSD. Thus it seems more reasonable to have config num.replica.move.threads that is shared across all log directories. Currently all Kafka configs, including num.recovery.threads.per.data.dir, defaults to a constant value instead of relying on values of configs. This it will be a bit weird if the config name itself is not per log dir but its default value is per dir. And it will also make both code and user documentation a bit more complicated because currently all configs, including num.recovery.threads.per.data.dir, defaults to a constant value. The advantage of using a magic value is simplicity. To answer your question, I think 3 ReplicaMoveThreads can work with more than 2 log directories. Say there are 3 ReplicaMoveThreads and 4 log directories, each ReplicaMoveThread will check if there is any replica waiting for movement, finish movement of this replica, and check again. Is there any concern with this approach? I have chosen the magic value 3 because current default number of network threads is 3. We can also set it to 8 which is the default number of io threads. Would there be any performance concern with using 8 threads by default? > > 20. Should we support canceling the movement across log dirs? I was > thinking this can be achieved with a ChangeReplicaDirRequest with dir = > any. > As of current KIP user can cancel movement across log directories by first sending DescribeDirsRequest, figure out the source directory of those replicas that are being moved, and then send ChangeReplicaDirRequest to move replica to the source log directory. But "any" seems like an easier and reasonable approach to cancel replica movement. I just added it to the KIP. > > Jun > > > On Mon, Mar 13, 2017 at 3:32 PM, Dong Lin <lindon...@gmail.com> wrote: > > > Hey Jun, > > > > Thanks much for your detailed comments. Please see my reply below. > > > > On Mon, Mar 13, 2017 at 9:09 AM, Jun Rao <j...@confluent.io> wrote: > > > > > Hi, Dong, > > > > > > Thanks for the updated KIP. Some more comments below. > > > > > > 10. For the .move log, do we perform any segment deletion (based on > > > retention) or log cleaning (if a compacted topic)? Or do we only enable > > > that after the swap? > > > > > > 11. kafka-reassign-partitions.sh > > > 11.1 If all reassigned replicas are in the current broker and only the > > log > > > directories have changed, we can probably optimize the tool to not > > trigger > > > partition reassignment through the controller and only > > > send ChangeReplicaDirRequest. > > > > > > > Yes, the reassignment script should not create the reassignment znode if > no > > replicas are not be moved between brokers. This falls into the "How to > move > > replica between log directories on the same broker" of the Proposed > Change > > section. > > > > > > > 11.2 If ChangeReplicaDirRequest specifies a replica that's not created > > yet, > > > could the broker just remember that in memory and create the replica > when > > > the creation is requested? This way, when doing cluster expansion, we > can > > > make sure that the new replicas on the new brokers are created in the > > right > > > log directory in the first place. We can also avoid the tool having to > > keep > > > issuing ChangeReplicaDirRequest in response to > > > ReplicaNotAvailableException. > > > > > > > I am concerned that the ChangeReplicaDirRequest would be lost if broker > > restarts after it sends ChangeReplicaDirResponse but before it receives > > LeaderAndIsrRequest. In this case, the user will receive success when > they > > initiate replica reassignment, but replica reassignment will never > complete > > when they verify the reassignment later. This would be confusing to user. > > > > There are three different approaches to this problem if broker has not > > created replica yet after it receives ChangeReplicaDirResquest: > > > > 1) Broker immediately replies to user with ReplicaNotAvailableException > and > > user can decide to retry again later. The advantage of this solution is > > that the broker logic is very simple and the reassignment script logic > also > > seems straightforward. The disadvantage is that user script has to retry. > > But it seems fine - we can set interval between retries to be 0.5 sec so > > that broker want be bombarded by those requests. This is the solution > > chosen in the current KIP. > > > > 2) Broker can put ChangeReplicaDirRequest in a purgatory with timeout and > > replies to user after the replica has been created. I didn't choose this > in > > the interest of keeping broker logic simpler. > > > > 3) Broker can remember that by making a mark in the disk, e.g. create > > topicPartition.tomove directory in the destination log directory. This > mark > > will be persisted across broker restart. This is the first idea I had > but I > > replaced it with solution 1) in the interest of keeping broker simple. > > > > It seems that solution 1) is the simplest one that works. But I am OK to > > switch to the other two solutions if we don't want the retry logic. What > do > > you think? > > > > > > 11.3 Do we need an option in the tool to specify intra.broker. > > > throttled.rate? > > > > > > > I don't find it useful to add this option to > kafka-reassign-partitions.sh. > > The reason we have the option "--throttle" in the script to throttle > > replication rate is that we usually want higher quota to fix an offline > > replica to get out of URP. But we are OK to have a lower quota if we are > > moving replica only to balance the cluster. Thus it is common for SRE to > > use different quota when using kafka-reassign-partitions.sh to move > replica > > between brokers. > > > > However, the only reason for moving replica between log directories of > the > > same broker is to balance cluster resource. Thus the option to > > specify intra.broker.throttled.rate in the tool is not that useful. I am > > inclined not to add this option to keep this tool's usage simpler. > > > > > > > > > > 12. DescribeDirsRequest > > > 12.1 In other requests like CreateTopicRequest, we return an empty list > > in > > > the response for an empty input list. If the input list is null, we > > return > > > everything. We should probably follow the same convention here. > > > > > > > Thanks. I wasn't aware of this convention. I have change > > DescribeDirsRequest so that "null" indicates "all". > > > > > > > 12.2 Do we need the topics field? Since the request is about log dirs, > it > > > makes sense to specify the log dirs. But it's weird to specify topics. > > > > > > > The topics field is not necessary. But it is useful to reduce the > response > > size in case user are only interested in the status of a few topics. For > > example, user may have initiated the reassignment of a given replica from > > one log directory to another log directory on the same broker, and the > user > > only wants to check the status of this given partition by looking > > at DescribeDirsResponse. Thus this field is useful. > > > > I am not sure if it is weird to call this request DescribeDirsRequest. > The > > response is a map from log directory to information to some partitions on > > the log directory. Do you think we need to change the name of the > request? > > > > > > > 12.3 DescribeDirsResponsePartition: Should we include firstOffset and > > > nextOffset in the response? That could be useful to track the progress > of > > > the movement. > > > > > > > Yeah good point. I agree it is useful to include logEndOffset in the > > response. According to Log.scala doc the logEndOffset is equivalent to > the > > nextOffset. User can track progress by checking the difference between > > logEndOffset of the given partition in the source and destination log > > directories. I have added logEndOffset to the > DescribeDirsResponsePartition > > in the KIP. > > > > But it seems that we don't need firstOffset in the response. Do you think > > firstOffset is still needed? > > > > > > > > > > 13. ChangeReplicaDirResponse: Do we need error code at both levels? > > > > > > > My bad. It is not needed. I have removed request level error code. I also > > added ChangeReplicaDirRequestTopic and ChangeReplicaDirResponseTopic to > > reduce duplication of the "topic" string in the request and response. > > > > > > > > > > 14. num.replica.move.threads: Does it default to # log dirs? > > > > > > > No. It doesn't. I expect default number to be set to a conservative value > > such as 3. It may be surprising to user if the number of threads increase > > just because they have assigned more log directories to Kafka broker. > > > > It seems that the number of replica move threads doesn't have to depend > on > > the number of log directories. It is possible to have one thread that > moves > > replicas across all log directories. On the other hand we can have > multiple > > threads to move replicas to the same log directory. For example, if > broker > > uses SSD, the CPU instead of disk IO may be the replica move bottleneck > and > > it will be faster to move replicas using multiple threads per log > > directory. > > > > > > > > > > Thanks, > > > > > > Jun > > > > > > On Thu, Mar 9, 2017 at 7:04 PM, Dong Lin <lindon...@gmail.com> wrote: > > > > > > > I just made one correction in the KIP. If broker receives > > > > ChangeReplicaDirRequest and the replica hasn't been created there, > the > > > > broker will respond ReplicaNotAvailableException. > > > > The kafka-reassignemnt-partitions.sh will need to re-send > > > > ChangeReplicaDirRequest in this case in order to wait for controller > to > > > > send LeaderAndIsrRequest to broker. The previous approach of creating > > an > > > > empty directory seems hacky. > > > > > > > > > > > > > > > > > > > > On Thu, Mar 9, 2017 at 6:33 PM, Dong Lin <lindon...@gmail.com> > wrote: > > > > > > > > > Hey Jun, > > > > > > > > > > Thanks for your comments! I have updated the KIP to address your > > > > comments. > > > > > Please see my reply inline. > > > > > > > > > > Can you let me know if the latest KIP has addressed your comments? > > > > > > > > > > On Wed, Mar 8, 2017 at 9:56 PM, Jun Rao <j...@confluent.io> wrote: > > > > > > > > > >> Hi, Dong, > > > > >> > > > > >> Thanks for the reply. > > > > >> > > > > >> 1.3 So the thread gets the lock, checks if caught up and releases > > the > > > > lock > > > > >> if not? Then, in the case when there is continuous incoming data, > > the > > > > >> thread may never get a chance to swap. One way to address this is > > when > > > > the > > > > >> thread is getting really close in catching up, just hold onto the > > lock > > > > >> until the thread fully catches up. > > > > >> > > > > > > > > > > Yes, that was my original solution. I see your point that the lock > > may > > > > not > > > > > be fairly assigned to ReplicaMoveThread and RequestHandlerThread > when > > > > there > > > > > is frequent incoming requets. You solution should address the > problem > > > > and I > > > > > have updated the KIP to use it. > > > > > > > > > > > > > > >> > > > > >> 2.3 So, you are saying that the partition reassignment tool can > > first > > > > send > > > > >> a ChangeReplicaDirRequest to relevant brokers to establish the log > > dir > > > > for > > > > >> replicas not created yet, then trigger the partition movement > across > > > > >> brokers through the controller? That's actually a good idea. Then, > > we > > > > can > > > > >> just leave LeaderAndIsrRequest as it is. > > > > > > > > > > > > > > > Yes, that is what I plan to do. If broker receives a > > > > > ChangeReplicaDirRequest while it is not leader or follower of the > > > > > partition, the broker will create an empty Log instance (i.e. a > > > directory > > > > > named topicPartition) in the destination log directory so that the > > > > replica > > > > > will be placed there when broker receives LeaderAndIsrRequest from > > the > > > > > broker. The broker should clean up empty those Log instances on > > startup > > > > > just in case a ChangeReplicaDirRequest was mistakenly sent to a > > broker > > > > that > > > > > was not meant to be follower/leader of the partition.. > > > > > > > > > > > > > > >> Another thing related to > > > > >> ChangeReplicaDirRequest. > > > > >> Since this request may take long to complete, I am not sure if we > > > should > > > > >> wait for the movement to complete before respond. While waiting > for > > > the > > > > >> movement to complete, the idle connection may be killed or the > > client > > > > may > > > > >> be gone already. An alternative is to return immediately and add a > > new > > > > >> request like CheckReplicaDirRequest to see if the movement has > > > > completed. > > > > >> The tool can take advantage of that to check the status. > > > > >> > > > > > > > > > > I agree with your concern and solution. We need request to query > the > > > > > partition -> log_directory mapping on the broker. I have updated > the > > > KIP > > > > to > > > > > remove need for ChangeReplicaDirRequestPurgatory. > > > > > Instead, kafka-reassignemnt-partitions.sh will send > > > DescribeDirsRequest > > > > > to brokers when user wants to verify the partition assignment. > Since > > we > > > > > need this DescribeDirsRequest anyway, we can also use this request > to > > > > > expose stats like the individual log size instead of using JMX. One > > > > > drawback of using JMX is that user has to manage the JMX port and > > > related > > > > > credentials if they haven't already done this, which is the case at > > > > > LinkedIn. > > > > > > > > > > > > > > >> Thanks, > > > > >> > > > > >> Jun > > > > >> > > > > >> > > > > >> > > > > >> On Wed, Mar 8, 2017 at 6:21 PM, Dong Lin <lindon...@gmail.com> > > wrote: > > > > >> > > > > >> > Hey Jun, > > > > >> > > > > > >> > Thanks for the detailed explanation. I will use the separate > > thread > > > > >> pool to > > > > >> > move replica between log directories. I will let you know when > the > > > KIP > > > > >> has > > > > >> > been updated to use a separate thread pool. > > > > >> > > > > > >> > Here is my response to your other questions: > > > > >> > > > > > >> > 1.3 My idea is that the ReplicaMoveThread that moves data should > > get > > > > the > > > > >> > lock before checking whether the replica in the destination log > > > > >> directory > > > > >> > has caught up. If the new replica has caught up, then the > > > > >> ReplicaMoveThread > > > > >> > should swaps the replica while it is still holding the lock. The > > > > >> > ReplicaFetcherThread or RequestHandlerThread will not be able to > > > > append > > > > >> > data to the replica in the source replica during this period > > because > > > > >> they > > > > >> > can not get the lock. Does this address the problem? > > > > >> > > > > > >> > 2.3 I get your point that we want to keep controller simpler. If > > > admin > > > > >> tool > > > > >> > can send ChangeReplicaDirRequest to move data within a broker, > > then > > > > >> > controller probably doesn't even need to include log directory > > path > > > in > > > > >> the > > > > >> > LeaderAndIsrRequest. How about this: controller will only deal > > with > > > > >> > reassignment across brokers as it does now. If user specified > > > > >> destination > > > > >> > replica for any disk, the admin tool will send > > > ChangeReplicaDirRequest > > > > >> and > > > > >> > wait for response from broker to confirm that all replicas have > > been > > > > >> moved > > > > >> > to the destination log direcotry. The broker will put > > > > >> > ChangeReplicaDirRequset in a purgatory and respond either when > the > > > > >> movement > > > > >> > is completed or when the request has timed-out. > > > > >> > > > > > >> > 4. I agree that we can expose these metrics via JMX. But I am > not > > > sure > > > > >> if > > > > >> > it can be obtained easily with good performance using either > > > existing > > > > >> tools > > > > >> > or new script in kafka. I will ask SREs for their opinion. > > > > >> > > > > > >> > Thanks, > > > > >> > Dong > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > On Wed, Mar 8, 2017 at 1:24 PM, Jun Rao <j...@confluent.io> > wrote: > > > > >> > > > > > >> > > Hi, Dong, > > > > >> > > > > > > >> > > Thanks for the updated KIP. A few more comments below. > > > > >> > > > > > > >> > > 1.1 and 1.2: I am still not sure there is enough benefit of > > > reusing > > > > >> > > ReplicaFetchThread > > > > >> > > to move data across disks. > > > > >> > > (a) A big part of ReplicaFetchThread is to deal with issuing > and > > > > >> tracking > > > > >> > > fetch requests. So, it doesn't feel that we get much from > > reusing > > > > >> > > ReplicaFetchThread > > > > >> > > only to disable the fetching part. > > > > >> > > (b) The leader replica has no ReplicaFetchThread to start > with. > > It > > > > >> feels > > > > >> > > weird to start one just for intra broker data movement. > > > > >> > > (c) The ReplicaFetchThread is per broker. Intuitively, the > > number > > > of > > > > >> > > threads doing intra broker data movement should be related to > > the > > > > >> number > > > > >> > of > > > > >> > > disks in the broker, not the number of brokers in the cluster. > > > > >> > > (d) If the destination disk fails, we want to stop the intra > > > broker > > > > >> data > > > > >> > > movement, but want to continue inter broker replication. So, > > > > >> logically, > > > > >> > it > > > > >> > > seems it's better to separate out the two. > > > > >> > > (e) I am also not sure if we should reuse the existing > > throttling > > > > for > > > > >> > > replication. It's designed to handle traffic across brokers > and > > > the > > > > >> > > delaying is done in the fetch request. So, if we are not doing > > > > >> > > fetching in ReplicaFetchThread, > > > > >> > > I am not sure the existing throttling is effective. Also, when > > > > >> specifying > > > > >> > > the throttling of moving data across disks, it seems the user > > > > >> shouldn't > > > > >> > > care about whether a replica is a leader or a follower. > Reusing > > > the > > > > >> > > existing throttling config name will be awkward in this > regard. > > > > >> > > (f) It seems it's simpler and more consistent to use a > separate > > > > thread > > > > >> > pool > > > > >> > > for local data movement (for both leader and follower > replicas). > > > > This > > > > >> > > process can then be configured (e.g. number of threads, etc) > and > > > > >> > throttled > > > > >> > > independently. > > > > >> > > > > > > >> > > 1.3 Yes, we will need some synchronization there. So, if the > > > > movement > > > > >> > > thread catches up, gets the lock to do the swap, but realizes > > that > > > > new > > > > >> > data > > > > >> > > is added, it has to continue catching up while holding the > lock? > > > > >> > > > > > > >> > > 2.3 The benefit of including the desired log directory in > > > > >> > > LeaderAndIsrRequest > > > > >> > > during partition reassignment is that the controller doesn't > > need > > > to > > > > >> > track > > > > >> > > the progress for disk movement. So, you don't need the > > additional > > > > >> > > BrokerDirStateUpdateRequest. Then the controller never needs > to > > > > issue > > > > >> > > ChangeReplicaDirRequest. > > > > >> > > Only the admin tool will issue ChangeReplicaDirRequest to move > > > data > > > > >> > within > > > > >> > > a broker. I agree that this makes LeaderAndIsrRequest more > > > > >> complicated, > > > > >> > but > > > > >> > > that seems simpler than changing the controller to track > > > additional > > > > >> > states > > > > >> > > during partition reassignment. > > > > >> > > > > > > >> > > 4. We want to make a decision on how to expose the stats. So > > far, > > > we > > > > >> are > > > > >> > > exposing stats like the individual log size as JMX. So, one > way > > is > > > > to > > > > >> > just > > > > >> > > add new jmx to expose the log directory of individual > replicas. > > > > >> > > > > > > >> > > Thanks, > > > > >> > > > > > > >> > > Jun > > > > >> > > > > > > >> > > > > > > >> > > On Thu, Mar 2, 2017 at 11:18 PM, Dong Lin < > lindon...@gmail.com> > > > > >> wrote: > > > > >> > > > > > > >> > > > Hey Jun, > > > > >> > > > > > > > >> > > > Thanks for all the comments! Please see my answer below. I > > have > > > > >> updated > > > > >> > > the > > > > >> > > > KIP to address most of the questions and make the KIP easier > > to > > > > >> > > understand. > > > > >> > > > > > > > >> > > > Thanks, > > > > >> > > > Dong > > > > >> > > > > > > > >> > > > On Thu, Mar 2, 2017 at 9:35 AM, Jun Rao <j...@confluent.io> > > > wrote: > > > > >> > > > > > > > >> > > > > Hi, Dong, > > > > >> > > > > > > > > >> > > > > Thanks for the KIP. A few comments below. > > > > >> > > > > > > > > >> > > > > 1. For moving data across directories > > > > >> > > > > 1.1 I am not sure why we want to use ReplicaFetcherThread > to > > > > move > > > > >> > data > > > > >> > > > > around in the leader. ReplicaFetchThread fetches data from > > > > socket. > > > > >> > For > > > > >> > > > > moving data locally, it seems that we want to avoid the > > socket > > > > >> > > overhead. > > > > >> > > > > > > > > >> > > > > > > > >> > > > The purpose of using ReplicaFetchThread is to re-use > existing > > > > thread > > > > >> > > > instead of creating more threads and make our thread model > > more > > > > >> > complex. > > > > >> > > It > > > > >> > > > seems like a nature choice for copying data between disks > > since > > > it > > > > >> is > > > > >> > > > similar to copying data between brokers. Another reason is > > that > > > if > > > > >> the > > > > >> > > > replica to be moved is a follower, we don't need lock to > swap > > > > >> replicas > > > > >> > > when > > > > >> > > > destination replica has caught up, since the same thread > which > > > is > > > > >> > > fetching > > > > >> > > > data from leader will swap the replica. > > > > >> > > > > > > > >> > > > The ReplicaFetchThread will not incur socket overhead while > > > > copying > > > > >> > data > > > > >> > > > between disks. It will read directly from source disk (as we > > do > > > > when > > > > >> > > > processing FetchRequest) and write to destination disk (as > we > > do > > > > >> when > > > > >> > > > processing ProduceRequest). > > > > >> > > > > > > > >> > > > > > > > >> > > > > 1.2 I am also not sure about moving data in the > > > > >> ReplicaFetcherThread > > > > >> > in > > > > >> > > > the > > > > >> > > > > follower. For example, I am not sure setting > > > > >> replica.fetch.max.wait > > > > >> > to > > > > >> > > 0 > > > > >> > > > > is ideal. It may not always be effective since a fetch > > > request > > > > in > > > > >> > the > > > > >> > > > > ReplicaFetcherThread could be arbitrarily delayed due to > > > > >> replication > > > > >> > > > > throttling on the leader. In general, the data movement > > logic > > > > >> across > > > > >> > > > disks > > > > >> > > > > seems different from that in ReplicaFetcherThread. So, I > am > > > not > > > > >> sure > > > > >> > > why > > > > >> > > > > they need to be coupled. > > > > >> > > > > > > > > >> > > > > > > > >> > > > While it may not be the most efficient way to copy data > > between > > > > >> local > > > > >> > > > disks, it will be at least as efficient as copying data from > > > > leader > > > > >> to > > > > >> > > the > > > > >> > > > destination disk. The expected goal of KIP-113 is to enable > > data > > > > >> > movement > > > > >> > > > between disks with no less efficiency than what we do now > when > > > > >> moving > > > > >> > > data > > > > >> > > > between brokers. I think we can optimize its performance > using > > > > >> separate > > > > >> > > > thread if the performance is not good enough. > > > > >> > > > > > > > >> > > > > > > > >> > > > > 1.3 Could you add a bit more details on how we swap the > > > replicas > > > > >> when > > > > >> > > the > > > > >> > > > > new ones are fully caught up? For example, what happens > when > > > the > > > > >> new > > > > >> > > > > replica in the new log directory is caught up, but when we > > > want > > > > >> to do > > > > >> > > the > > > > >> > > > > swap, some new data has arrived? > > > > >> > > > > > > > > >> > > > > > > > >> > > > If the replica is a leader, then ReplicaFetcherThread will > > > perform > > > > >> the > > > > >> > > > replacement. Proper lock is needed to prevent > > > KafkaRequestHandler > > > > >> from > > > > >> > > > appending data to the topicPartition.log on the source disks > > > > before > > > > >> > this > > > > >> > > > replacement is completed by ReplicaFetcherThread. > > > > >> > > > > > > > >> > > > If the replica is a follower, because the same > > > ReplicaFetchThread > > > > >> which > > > > >> > > > fetches data from leader will also swap the replica , no > lock > > is > > > > >> > needed. > > > > >> > > > > > > > >> > > > I have updated the KIP to specify both more explicitly. > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > >> > > > > 1.4 Do we need to do the .move at the log segment level or > > > could > > > > >> we > > > > >> > > just > > > > >> > > > do > > > > >> > > > > that at the replica directory level? Renaming just a > > directory > > > > is > > > > >> > much > > > > >> > > > > faster than renaming the log segments. > > > > >> > > > > > > > > >> > > > > > > > >> > > > Great point. I have updated the KIP to rename the log > > directory > > > > >> > instead. > > > > >> > > > > > > > >> > > > > > > > >> > > > > 1.5 Could you also describe a bit what happens when either > > the > > > > >> source > > > > >> > > or > > > > >> > > > > the target log directory fails while the data moving is in > > > > >> progress? > > > > >> > > > > > > > > >> > > > > > > > >> > > > If source log directory fails, then the replica movement > will > > > stop > > > > >> and > > > > >> > > the > > > > >> > > > source replica is marked offline. If destination log > directory > > > > >> fails, > > > > >> > > then > > > > >> > > > the replica movement will stop. I have updated the KIP to > > > clarify > > > > >> this. > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > > >> > > > > 2. For partition reassignment. > > > > >> > > > > 2.1 I am not sure if the controller can block on > > > > >> > > ChangeReplicaDirRequest. > > > > >> > > > > Data movement may take a long time to complete. If there > is > > an > > > > >> > > > outstanding > > > > >> > > > > request from the controller to a broker, that broker won't > > be > > > > >> able to > > > > >> > > > > process any new request from the controller. So if another > > > event > > > > >> > (e.g. > > > > >> > > > > broker failure) happens when the data movement is in > > progress, > > > > >> > > subsequent > > > > >> > > > > LeaderAnIsrRequest will be delayed. > > > > >> > > > > > > > > >> > > > > > > > >> > > > Yeah good point. I missed the fact that there is be only one > > > > >> inflight > > > > >> > > > request from controller to broker. > > > > >> > > > > > > > >> > > > How about I add a request, e.g. BrokerDirStateUpdateRequest, > > > which > > > > >> maps > > > > >> > > > topicPartition to log directory and can be sent from broker > to > > > > >> > controller > > > > >> > > > to indicate completion? > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > >> > > > > 2.2 in the KIP, the partition reassignment tool is also > used > > > for > > > > >> > cases > > > > >> > > > > where an admin just wants to balance the existing data > > across > > > > log > > > > >> > > > > directories in the broker. In this case, it seems that > it's > > > over > > > > >> > > killing > > > > >> > > > to > > > > >> > > > > have the process go through the controller. A simpler > > approach > > > > is > > > > >> to > > > > >> > > > issue > > > > >> > > > > an RPC request to the broker directly. > > > > >> > > > > > > > > >> > > > > > > > >> > > > I agree we can optimize this case. It is just that we have > to > > > add > > > > >> new > > > > >> > > logic > > > > >> > > > or code path to handle a scenario that is already covered by > > the > > > > >> more > > > > >> > > > complicated scenario. I will add it to the KIP. > > > > >> > > > > > > > >> > > > > > > > >> > > > > 2.3 When using the partition reassignment tool to move > > > replicas > > > > >> > across > > > > >> > > > > brokers, it make sense to be able to specify the log > > directory > > > > of > > > > >> the > > > > >> > > > newly > > > > >> > > > > created replicas. The KIP does that in two separate > requests > > > > >> > > > > ChangeReplicaDirRequest and LeaderAndIsrRequest, and > tracks > > > the > > > > >> > > progress > > > > >> > > > of > > > > >> > > > > each independently. An alternative is to do that just in > > > > >> > > > > LeaderAndIsrRequest. > > > > >> > > > > That way, the new replicas will be created in the right > log > > > dir > > > > in > > > > >> > the > > > > >> > > > > first place and the controller just needs to track the > > > progress > > > > of > > > > >> > > > > partition reassignment in the current way. > > > > >> > > > > > > > > >> > > > > > > > >> > > > I agree it is better to use one request instead of two to > > > request > > > > >> > replica > > > > >> > > > movement between disks. But I think the performance > advantage > > of > > > > >> doing > > > > >> > so > > > > >> > > > is negligible because we trigger replica assignment much > less > > > than > > > > >> all > > > > >> > > > other kinds of events in the Kafka cluster. I am not sure > that > > > the > > > > >> > > benefit > > > > >> > > > of doing this is worth the effort to add an optional string > > > field > > > > in > > > > >> > the > > > > >> > > > LeaderAndIsrRequest. Also if we add this optional field in > the > > > > >> > > > LeaderAndIsrRequest, we probably want to remove > > > > >> ChangeReplicaDirRequest > > > > >> > > to > > > > >> > > > avoid having two requests doing the same thing. But it means > > > user > > > > >> > script > > > > >> > > > can not send request directly to the broker to trigger > replica > > > > >> movement > > > > >> > > > between log directories. > > > > >> > > > > > > > >> > > > I will do it if you are strong about this optimzation. > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > > >> > > > > 3. /admin/reassign_partitions: Including the log dir in > > every > > > > >> replica > > > > >> > > may > > > > >> > > > > not be efficient. We could include a list of log > directories > > > and > > > > >> > > > reference > > > > >> > > > > the index of the log directory in each replica. > > > > >> > > > > > > > > >> > > > > > > > >> > > > Good point. I have updated the KIP to use this solution. > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > > >> > > > > 4. DescribeDirsRequest: The stats in the request are > already > > > > >> > available > > > > >> > > > from > > > > >> > > > > JMX. Do we need the new request? > > > > >> > > > > > > > > >> > > > > > > > >> > > > Does JMX also include the state (i.e. offline or online) of > > each > > > > log > > > > >> > > > directory and the log directory of each replica? If not, > then > > > > maybe > > > > >> we > > > > >> > > > still need DescribeDirsRequest? > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > > >> > > > > 5. We want to be consistent on ChangeReplicaDirRequest vs > > > > >> > > > > ChangeReplicaRequest. > > > > >> > > > > > > > > >> > > > > > > > >> > > > I think ChangeReplicaRequest and ChangeReplicaResponse is my > > > typo. > > > > >> > Sorry, > > > > >> > > > they are fixed now. > > > > >> > > > > > > > >> > > > > > > > >> > > > > > > > > >> > > > > Thanks, > > > > >> > > > > > > > > >> > > > > Jun > > > > >> > > > > > > > > >> > > > > > > > > >> > > > > On Fri, Feb 3, 2017 at 6:19 PM, Dong Lin < > > lindon...@gmail.com > > > > > > > > >> > wrote: > > > > >> > > > > > > > > >> > > > > > Hey ALexey, > > > > >> > > > > > > > > > >> > > > > > Thanks for all the comments! > > > > >> > > > > > > > > > >> > > > > > I have updated the KIP to specify how we enforce quota. > I > > > also > > > > >> > > updated > > > > >> > > > > the > > > > >> > > > > > "The thread model and broker logic for moving replica > data > > > > >> between > > > > >> > > log > > > > >> > > > > > directories" to make it easier to read. You can find the > > > exact > > > > >> > change > > > > >> > > > > here > > > > >> > > > > > <https://cwiki.apache.org/conf > > > luence/pages/diffpagesbyversio > > > > >> > > > > > n.action?pageId=67638408&selec > > > tedPageVersions=5&selectedPage > > > > >> > > > Versions=6>. > > > > >> > > > > > The idea is to use the same replication quota mechanism > > > > >> introduced > > > > >> > in > > > > >> > > > > > KIP-73. > > > > >> > > > > > > > > > >> > > > > > Thanks, > > > > >> > > > > > Dong > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > On Wed, Feb 1, 2017 at 2:16 AM, Alexey Ozeritsky < > > > > >> > > aozerit...@yandex.ru > > > > >> > > > > > > > > >> > > > > > wrote: > > > > >> > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > 24.01.2017, 22:03, "Dong Lin" <lindon...@gmail.com>: > > > > >> > > > > > > > Hey Alexey, > > > > >> > > > > > > > > > > > >> > > > > > > > Thanks. I think we agreed that the suggested > solution > > > > >> doesn't > > > > >> > > work > > > > >> > > > in > > > > >> > > > > > > > general for kafka users. To answer your questions: > > > > >> > > > > > > > > > > > >> > > > > > > > 1. I agree we need quota to rate limit replica > > movement > > > > >> when a > > > > >> > > > broker > > > > >> > > > > > is > > > > >> > > > > > > > moving a "leader" replica. I will come up with > > solution, > > > > >> > probably > > > > >> > > > > > re-use > > > > >> > > > > > > > the config of replication quota introduced in > KIP-73. > > > > >> > > > > > > > > > > > >> > > > > > > > 2. Good point. I agree that this is a problem in > > > general. > > > > >> If is > > > > >> > > no > > > > >> > > > > new > > > > >> > > > > > > data > > > > >> > > > > > > > on that broker, with current default value of > > > > >> > > > > > replica.fetch.wait.max.ms > > > > >> > > > > > > > and replica.fetch.max.bytes, the replica will be > moved > > > at > > > > >> only > > > > >> > 2 > > > > >> > > > MBps > > > > >> > > > > > > > throughput. I think the solution is for broker to > set > > > > >> > > > > > > > replica.fetch.wait.max.ms to 0 in its FetchRequest > if > > > the > > > > >> > > > > > corresponding > > > > >> > > > > > > > ReplicaFetcherThread needs to move some replica to > > > another > > > > >> > disk. > > > > >> > > > > > > > > > > > >> > > > > > > > 3. I have updated the KIP to mention that the read > > size > > > > of a > > > > >> > > given > > > > >> > > > > > > > partition is configured using > replica.fetch.max.bytes > > > when > > > > >> we > > > > >> > > move > > > > >> > > > > > > replicas > > > > >> > > > > > > > between disks. > > > > >> > > > > > > > > > > > >> > > > > > > > Please see this > > > > >> > > > > > > > <https://cwiki.apache.org/conf > > > > >> luence/pages/diffpagesbyversio > > > > >> > > > n.action > > > > >> > > > > ? > > > > >> > > > > > > pageId=67638408&selectedPageVe > > > > rsions=4&selectedPageVersions= > > > > >> 5> > > > > >> > > > > > > > for the change of the KIP. I will come up with a > > > solution > > > > to > > > > >> > > > throttle > > > > >> > > > > > > > replica movement when a broker is moving a "leader" > > > > replica. > > > > >> > > > > > > > > > > >> > > > > > > Thanks. It looks great. > > > > >> > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > On Tue, Jan 24, 2017 at 3:30 AM, Alexey Ozeritsky < > > > > >> > > > > > aozerit...@yandex.ru> > > > > >> > > > > > > > wrote: > > > > >> > > > > > > > > > > > >> > > > > > > >> 23.01.2017, 22:11, "Dong Lin" < > lindon...@gmail.com > > >: > > > > >> > > > > > > >> > Thanks. Please see my comment inline. > > > > >> > > > > > > >> > > > > > >> > > > > > > >> > On Mon, Jan 23, 2017 at 6:45 AM, Alexey > Ozeritsky > > < > > > > >> > > > > > > aozerit...@yandex.ru> > > > > >> > > > > > > >> > wrote: > > > > >> > > > > > > >> > > > > > >> > > > > > > >> >> 13.01.2017, 22:29, "Dong Lin" < > > lindon...@gmail.com > > > >: > > > > >> > > > > > > >> >> > Hey Alexey, > > > > >> > > > > > > >> >> > > > > > >> > > > > > > >> >> > Thanks for your review and the alternative > > > > approach. > > > > >> > Here > > > > >> > > is > > > > >> > > > > my > > > > >> > > > > > > >> >> > understanding of your patch. kafka's > background > > > > >> threads > > > > >> > > are > > > > >> > > > > used > > > > >> > > > > > > to > > > > >> > > > > > > >> move > > > > >> > > > > > > >> >> > data between replicas. When data movement is > > > > >> triggered, > > > > >> > > the > > > > >> > > > > log > > > > >> > > > > > > will > > > > >> > > > > > > >> be > > > > >> > > > > > > >> >> > rolled and the new logs will be put in the > new > > > > >> > directory, > > > > >> > > > and > > > > >> > > > > > > >> background > > > > >> > > > > > > >> >> > threads will move segment from old directory > to > > > new > > > > >> > > > directory. > > > > >> > > > > > > >> >> > > > > > >> > > > > > > >> >> > It is important to note that KIP-112 is > > intended > > > to > > > > >> work > > > > >> > > > with > > > > >> > > > > > > >> KIP-113 to > > > > >> > > > > > > >> >> > support JBOD. I think your solution is > > definitely > > > > >> > simpler > > > > >> > > > and > > > > >> > > > > > > better > > > > >> > > > > > > >> >> under > > > > >> > > > > > > >> >> > the current kafka implementation that a > broker > > > will > > > > >> fail > > > > >> > > if > > > > >> > > > > any > > > > >> > > > > > > disk > > > > >> > > > > > > >> >> fails. > > > > >> > > > > > > >> >> > But I am not sure if we want to allow broker > to > > > run > > > > >> with > > > > >> > > > > partial > > > > >> > > > > > > >> disks > > > > >> > > > > > > >> >> > failure. Let's say the a replica is being > moved > > > > from > > > > >> > > > > log_dir_old > > > > >> > > > > > > to > > > > >> > > > > > > >> >> > log_dir_new and then log_dir_old stops > working > > > due > > > > to > > > > >> > disk > > > > >> > > > > > > failure. > > > > >> > > > > > > >> How > > > > >> > > > > > > >> >> > would your existing patch handles it? To make > > the > > > > >> > > scenario a > > > > >> > > > > bit > > > > >> > > > > > > more > > > > >> > > > > > > >> >> > > > > >> > > > > > > >> >> We will lose log_dir_old. After broker restart > we > > > can > > > > >> read > > > > >> > > the > > > > >> > > > > > data > > > > >> > > > > > > >> from > > > > >> > > > > > > >> >> log_dir_new. > > > > >> > > > > > > >> > > > > > >> > > > > > > >> > No, you probably can't. This is because the > broker > > > > >> doesn't > > > > >> > > have > > > > >> > > > > > > *all* the > > > > >> > > > > > > >> > data for this partition. For example, say the > > broker > > > > has > > > > >> > > > > > > >> > partition_segement_1, partition_segment_50 and > > > > >> > > > > > partition_segment_100 > > > > >> > > > > > > on > > > > >> > > > > > > >> the > > > > >> > > > > > > >> > log_dir_old. partition_segment_100, which has > the > > > > latest > > > > >> > > data, > > > > >> > > > > has > > > > >> > > > > > > been > > > > >> > > > > > > >> > moved to log_dir_new, and the log_dir_old fails > > > before > > > > >> > > > > > > >> partition_segment_50 > > > > >> > > > > > > >> > and partition_segment_1 is moved to log_dir_new. > > > When > > > > >> > broker > > > > >> > > > > > > re-starts, > > > > >> > > > > > > >> it > > > > >> > > > > > > >> > won't have partition_segment_50. This causes > > problem > > > > if > > > > >> > > broker > > > > >> > > > is > > > > >> > > > > > > elected > > > > >> > > > > > > >> > leader and consumer wants to consume data in the > > > > >> > > > > > partition_segment_1. > > > > >> > > > > > > >> > > > > >> > > > > > > >> Right. > > > > >> > > > > > > >> > > > > >> > > > > > > >> > > > > > >> > > > > > > >> >> > complicated, let's say the broker is > shtudown, > > > > >> > > log_dir_old's > > > > >> > > > > > disk > > > > >> > > > > > > >> fails, > > > > >> > > > > > > >> >> > and the broker starts. In this case broker > > > doesn't > > > > >> even > > > > >> > > know > > > > >> > > > > if > > > > >> > > > > > > >> >> log_dir_new > > > > >> > > > > > > >> >> > has all the data needed for this replica. It > > > > becomes > > > > >> a > > > > >> > > > problem > > > > >> > > > > > if > > > > >> > > > > > > the > > > > >> > > > > > > >> >> > broker is elected leader of this partition in > > > this > > > > >> case. > > > > >> > > > > > > >> >> > > > > >> > > > > > > >> >> log_dir_new contains the most recent data so we > > > will > > > > >> lose > > > > >> > > the > > > > >> > > > > tail > > > > >> > > > > > > of > > > > >> > > > > > > >> >> partition. > > > > >> > > > > > > >> >> This is not a big problem for us because we > > already > > > > >> delete > > > > >> > > > tails > > > > >> > > > > > by > > > > >> > > > > > > >> hand > > > > >> > > > > > > >> >> (see https://issues.apache.org/jira > > > > /browse/KAFKA-1712 > > > > >> ). > > > > >> > > > > > > >> >> Also we dont use authomatic leader balancing > > > > >> > > > > > > >> (auto.leader.rebalance.enable=false), > > > > >> > > > > > > >> >> so this partition becomes the leader with a low > > > > >> > probability. > > > > >> > > > > > > >> >> I think my patch can be modified to prohibit > the > > > > >> selection > > > > >> > > of > > > > >> > > > > the > > > > >> > > > > > > >> leader > > > > >> > > > > > > >> >> until the partition does not move completely. > > > > >> > > > > > > >> > > > > > >> > > > > > > >> > I guess you are saying that you have deleted the > > > tails > > > > >> by > > > > >> > > hand > > > > >> > > > in > > > > >> > > > > > > your > > > > >> > > > > > > >> own > > > > >> > > > > > > >> > kafka branch. But KAFKA-1712 is not accepted > into > > > > Kafka > > > > >> > trunk > > > > >> > > > > and I > > > > >> > > > > > > am > > > > >> > > > > > > >> not > > > > >> > > > > > > >> > > > > >> > > > > > > >> No. We just modify segments mtime by cron job. > This > > > > works > > > > >> > with > > > > >> > > > > > vanilla > > > > >> > > > > > > >> kafka. > > > > >> > > > > > > >> > > > > >> > > > > > > >> > sure if it is the right solution. How would this > > > > >> solution > > > > >> > > > address > > > > >> > > > > > the > > > > >> > > > > > > >> > problem mentioned above? > > > > >> > > > > > > >> > > > > >> > > > > > > >> If you need only fresh data and if you remove old > > data > > > > by > > > > >> > hands > > > > >> > > > > this > > > > >> > > > > > is > > > > >> > > > > > > >> not a problem. But in general case > > > > >> > > > > > > >> this is a problem of course. > > > > >> > > > > > > >> > > > > >> > > > > > > >> > > > > > >> > > > > > > >> > BTW, I am not sure the solution mentioned in > > > > KAFKA-1712 > > > > >> is > > > > >> > > the > > > > >> > > > > > right > > > > >> > > > > > > way > > > > >> > > > > > > >> to > > > > >> > > > > > > >> > address its problem. Now that we have timestamp > in > > > the > > > > >> > > message > > > > >> > > > we > > > > >> > > > > > > can use > > > > >> > > > > > > >> > that to delete old segement instead of relying > on > > > the > > > > >> log > > > > >> > > > segment > > > > >> > > > > > > mtime. > > > > >> > > > > > > >> > Just some idea and we don't have to discuss this > > > > problem > > > > >> > > here. > > > > >> > > > > > > >> > > > > > >> > > > > > > >> >> > > > > > >> > > > > > > >> >> > The solution presented in the KIP attempts to > > > > handle > > > > >> it > > > > >> > by > > > > >> > > > > > > replacing > > > > >> > > > > > > >> >> > replica in an atomic version fashion after > the > > > log > > > > in > > > > >> > the > > > > >> > > > new > > > > >> > > > > > dir > > > > >> > > > > > > has > > > > >> > > > > > > >> >> fully > > > > >> > > > > > > >> >> > caught up with the log in the old dir. At at > > time > > > > the > > > > >> > log > > > > >> > > > can > > > > >> > > > > be > > > > >> > > > > > > >> >> considered > > > > >> > > > > > > >> >> > to exist on only one log directory. > > > > >> > > > > > > >> >> > > > > >> > > > > > > >> >> As I understand your solution does not cover > > > quotas. > > > > >> > > > > > > >> >> What happens if someone starts to transfer 100 > > > > >> partitions > > > > >> > ? > > > > >> > > > > > > >> > > > > > >> > > > > > > >> > Good point. Quota can be implemented in the > > future. > > > It > > > > >> is > > > > >> > > > > currently > > > > >> > > > > > > >> > mentioned as as a potential future improvement > in > > > > >> KIP-112 > > > > >> > > > > > > >> > <https://cwiki.apache.org/conf > > > > luence/display/KAFKA/KIP- > > > > >> > 112%3 > > > > >> > > > > > > >> A+Handle+disk+failure+for+JBOD>.Thanks > > > > >> > > > > > > >> > for the reminder. I will move it to KIP-113. > > > > >> > > > > > > >> > > > > > >> > > > > > > >> >> > If yes, it will read a ByteBufferMessageSet > > from > > > > >> > > > > > > topicPartition.log > > > > >> > > > > > > >> and > > > > >> > > > > > > >> >> append the message set to topicPartition.move > > > > >> > > > > > > >> >> > > > > >> > > > > > > >> >> i.e. processPartitionData will read data from > the > > > > >> > beginning > > > > >> > > of > > > > >> > > > > > > >> >> topicPartition.log? What is the read size? > > > > >> > > > > > > >> >> A ReplicaFetchThread reads many partitions so > if > > > one > > > > >> does > > > > >> > > some > > > > >> > > > > > > >> complicated > > > > >> > > > > > > >> >> work (= read a lot of data from disk) > everything > > > will > > > > >> slow > > > > >> > > > down. > > > > >> > > > > > > >> >> I think read size should not be very big. > > > > >> > > > > > > >> >> > > > > >> > > > > > > >> >> On the other hand at this point > > > > (processPartitionData) > > > > >> one > > > > >> > > can > > > > >> > > > > use > > > > >> > > > > > > only > > > > >> > > > > > > >> >> the new data (ByteBufferMessageSet from > > parameters) > > > > and > > > > >> > wait > > > > >> > > > > until > > > > >> > > > > > > >> >> (topicPartition.move.smallestOffset <= > > > > >> > > > > > > topicPartition.log.smallestOff > > > > >> > > > > > > >> set > > > > >> > > > > > > >> >> && topicPartition.log.largestOffset == > > > > >> > > > > > > topicPartition.log.largestOffs > > > > >> > > > > > > >> et). > > > > >> > > > > > > >> >> In this case the write speed to > > topicPartition.move > > > > and > > > > >> > > > > > > >> topicPartition.log > > > > >> > > > > > > >> >> will be the same so this will allow us to move > > many > > > > >> > > partitions > > > > >> > > > > to > > > > >> > > > > > > one > > > > >> > > > > > > >> disk. > > > > >> > > > > > > >> > > > > > >> > > > > > > >> > The read size of a given partition is configured > > > > >> > > > > > > >> > using replica.fetch.max.bytes, which is the same > > > size > > > > >> used > > > > >> > by > > > > >> > > > > > > >> FetchRequest > > > > >> > > > > > > >> > from follower to leader. If the broker is > moving a > > > > >> replica > > > > >> > > for > > > > >> > > > > > which > > > > >> > > > > > > it > > > > >> > > > > > > >> > > > > >> > > > > > > >> OK. Could you mention it in KIP? > > > > >> > > > > > > >> > > > > >> > > > > > > >> > acts as a follower, the disk write rate for > moving > > > > this > > > > >> > > replica > > > > >> > > > > is > > > > >> > > > > > at > > > > >> > > > > > > >> most > > > > >> > > > > > > >> > the rate it fetches from leader (assume it is > > > catching > > > > >> up > > > > >> > and > > > > >> > > > has > > > > >> > > > > > > >> > sufficient data to read from leader, which is > > > subject > > > > to > > > > >> > > > > > > round-trip-time > > > > >> > > > > > > >> > between itself and the leader. Thus this part if > > > > >> probably > > > > >> > > fine > > > > >> > > > > even > > > > >> > > > > > > >> without > > > > >> > > > > > > >> > quota. > > > > >> > > > > > > >> > > > > >> > > > > > > >> I think there are 2 problems > > > > >> > > > > > > >> 1. Without speed limiter this will not work good > > even > > > > for > > > > >> 1 > > > > >> > > > > > partition. > > > > >> > > > > > > In > > > > >> > > > > > > >> our production we had a problem so we did the > > throuput > > > > >> > limiter: > > > > >> > > > > > > >> https://github.com/resetius/ka > > > > >> fka/commit/cda31dadb2f135743bf > > > > >> > > > > > > >> 41083062927886c5ddce1#diff-ffa > > > > >> 8861e850121997a534ebdde2929c6R > > > > >> > > 713 > > > > >> > > > > > > >> > > > > >> > > > > > > >> 2. I dont understand how it will work in case of > big > > > > >> > > > > > > >> replica.fetch.wait.max.ms and partition with > > > irregular > > > > >> flow. > > > > >> > > > > > > >> For example someone could have > > > > replica.fetch.wait.max.ms > > > > >> > =10mi > > > > >> > > > nutes > > > > >> > > > > > and > > > > >> > > > > > > >> partition that has very high data flow from 12:00 > to > > > > 13:00 > > > > >> > and > > > > >> > > > zero > > > > >> > > > > > > flow > > > > >> > > > > > > >> otherwise. > > > > >> > > > > > > >> In this case processPartitionData could be called > > once > > > > per > > > > >> > > > > 10minutes > > > > >> > > > > > > so if > > > > >> > > > > > > >> we start data moving in 13:01 it will be finished > > next > > > > >> day. > > > > >> > > > > > > >> > > > > >> > > > > > > >> > > > > > >> > > > > > > >> > But ff the broker is moving a replica for which > it > > > > acts > > > > >> as > > > > >> > a > > > > >> > > > > > leader, > > > > >> > > > > > > as > > > > >> > > > > > > >> of > > > > >> > > > > > > >> > current KIP the broker will keep reading from > > > > >> log_dir_old > > > > >> > and > > > > >> > > > > > append > > > > >> > > > > > > to > > > > >> > > > > > > >> > log_dir_new without having to wait for > > > > round-trip-time. > > > > >> We > > > > >> > > > > probably > > > > >> > > > > > > need > > > > >> > > > > > > >> > quota for this in the future. > > > > >> > > > > > > >> > > > > > >> > > > > > > >> >> > > > > > >> > > > > > > >> >> > And to answer your question, yes > > > topicpartition.log > > > > >> > refers > > > > >> > > > to > > > > >> > > > > > > >> >> > topic-paritition/segment.log. > > > > >> > > > > > > >> >> > > > > > >> > > > > > > >> >> > Thanks, > > > > >> > > > > > > >> >> > Dong > > > > >> > > > > > > >> >> > > > > > >> > > > > > > >> >> > On Fri, Jan 13, 2017 at 4:12 AM, Alexey > > > Ozeritsky < > > > > >> > > > > > > >> aozerit...@yandex.ru> > > > > >> > > > > > > >> >> > wrote: > > > > >> > > > > > > >> >> > > > > > >> > > > > > > >> >> >> Hi, > > > > >> > > > > > > >> >> >> > > > > >> > > > > > > >> >> >> We have the similar solution that have been > > > > working > > > > >> in > > > > >> > > > > > production > > > > >> > > > > > > >> since > > > > >> > > > > > > >> >> >> 2014. You can see it here: > > > > >> > > https://github.com/resetius/ka > > > > >> > > > > > > >> >> >> fka/commit/20658593e246d218490 > > > > 6879defa2e763c4d413fb > > > > >> > > > > > > >> >> >> The idea is very simple > > > > >> > > > > > > >> >> >> 1. Disk balancer runs in a separate thread > > > inside > > > > >> > > scheduler > > > > >> > > > > > pool. > > > > >> > > > > > > >> >> >> 2. It does not touch empty partitions > > > > >> > > > > > > >> >> >> 3. Before it moves a partition it forcibly > > > creates > > > > >> new > > > > >> > > > > segment > > > > >> > > > > > > on a > > > > >> > > > > > > >> >> >> destination disk > > > > >> > > > > > > >> >> >> 4. It moves segment by segment from new to > > old. > > > > >> > > > > > > >> >> >> 5. Log class works with segments on both > disks > > > > >> > > > > > > >> >> >> > > > > >> > > > > > > >> >> >> Your approach seems too complicated, > moreover > > it > > > > >> means > > > > >> > > that > > > > >> > > > > you > > > > >> > > > > > > >> have to > > > > >> > > > > > > >> >> >> patch different components of the system > > > > >> > > > > > > >> >> >> Could you clarify what do you mean by > > > > >> > topicPartition.log? > > > > >> > > > Is > > > > >> > > > > it > > > > >> > > > > > > >> >> >> topic-paritition/segment.log ? > > > > >> > > > > > > >> >> >> > > > > >> > > > > > > >> >> >> 12.01.2017, 21:47, "Dong Lin" < > > > > lindon...@gmail.com > > > > >> >: > > > > >> > > > > > > >> >> >> > Hi all, > > > > >> > > > > > > >> >> >> > > > > > >> > > > > > > >> >> >> > We created KIP-113: Support replicas > > movement > > > > >> between > > > > >> > > log > > > > >> > > > > > > >> >> directories. > > > > >> > > > > > > >> >> >> > Please find the KIP wiki in the link > > > > >> > > > > > > >> >> >> > *https://cwiki.apache.org/conf > > > > >> > > > > luence/display/KAFKA/KIP-113% > > > > >> > > > > > > >> >> >> 3A+Support+replicas+movement+b > > > > >> etween+log+directories > > > > >> > > > > > > >> >> >> > <https://cwiki.apache.org/conf > > > > >> > > > > luence/display/KAFKA/KIP-113% > > > > >> > > > > > > >> >> >> 3A+Support+replicas+movement+ > > > > >> > between+log+directories>.* > > > > >> > > > > > > >> >> >> > > > > > >> > > > > > > >> >> >> > This KIP is related to KIP-112 > > > > >> > > > > > > >> >> >> > <https://cwiki.apache.org/conf > > > > >> > > > > luence/display/KAFKA/KIP-112% > > > > >> > > > > > > >> >> >> 3A+Handle+disk+failure+for+JBOD>: > > > > >> > > > > > > >> >> >> > Handle disk failure for JBOD. They are > > needed > > > in > > > > >> > order > > > > >> > > to > > > > >> > > > > > > support > > > > >> > > > > > > >> >> JBOD in > > > > >> > > > > > > >> >> >> > Kafka. Please help review the KIP. You > > > feedback > > > > is > > > > >> > > > > > appreciated! > > > > >> > > > > > > >> >> >> > > > > > >> > > > > > > >> >> >> > Thanks, > > > > >> > > > > > > >> >> >> > Dong > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > >> > > > > > > > > > > > > > > > > > > > >
