Mailing-List: contact dev-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@kafka.apache.org
MIME-Version: 1.0
In-Reply-To: <CAAaarBY4LEC+Wpwg50mrQ7zQQu_po7TQwDKujnq97z2fCdKRew@mail.gmail.com>
References: <CAAaarBbdZgzd_TkoR1e=3e-y0P17r3aYwprxsz5tkMTYE2ne7Q@mail.gmail.com>
 <1485372889.4045818.859516968.526900DB@webmail.messagingengine.com>
 <CAAaarBbG8Ug12v5JQMonhx78F+71R3QwgTo8psZ=NZcK_eYGYA@mail.gmail.com>
 <1485454559.147095.860587032.34ED4FF3@webmail.messagingengine.com>
 <CAAaarBYzSVYXH7SvFjHPHnLPmoHhM0igmoZTA_cgitJY9uDFrg@mail.gmail.com>
 <CANTWn9MVhMY59oDb3WZ6v8XP=twbY84ycL_q43VZuYNW4tHZtg@mail.gmail.com>
 <CAAaarBYKP8v79XbM998en2HDmfPJqfjXknu-mBdiBW773GBh7g@mail.gmail.com>
 <55CD9C37-7DDE-4984-8892-9296A52367F7@gmail.com> <CAAaarBZLs8in8GOgD89J7iaLfYaGwE34uUHnLkGLeuQ8TAaVOg@mail.gmail.com>
 <EA827A3C-EBC4-42B1-A187-6B9107A18861@gmail.com> <CAAaarBaqoKOOv0kp_GKuxHhrDvievFd9=4mZLVLbKCC+TibMdg@mail.gmail.com>
 <CAAaarBbQcig8e7fs5D3DV0GFbfHG7o9j7xBN__U8nUj_ZBa5cA@mail.gmail.com>
 <1486666986.161731.875963936.1F3D8FB9@webmail.messagingengine.com>
 <1486667065.161845.875971632.2A069636@webmail.messagingengine.com>
 <CAAaarBbZAOv2MJVNQ91nN2WpweEgsw09QotaF14NPmjK_BFhyA@mail.gmail.com>
 <1486683437.218609.876240160.4E98C449@webmail.messagingengine.com>
 <CAAaarBYrfdWDNHHSAqWvoJyuoSpU7uF0wuVDSjmURo=usZL2kg@mail.gmail.com>
 <CAAaarBaNTjAZfsiJQsMAWE=X47PNq2_A6QXHC5mNfk3RNVoALw@mail.gmail.com>
 <CAFc58G-8c71TzpB5twR1emeS4OhMHsH+iu-OBw=+kqx96R=DbA@mail.gmail.com>
 <CAAaarBaTRFa=RbXBKXuhBPxjuw2ifqUauW937=ZtBZ0muAvsAQ@mail.gmail.com>
 <CAFc58G-vzaYG-FOYUKVbp+SwqFnBm_UservRgi4nUPJ0icZa2A@mail.gmail.com>
 <CAAaarBY6s73NSCDHA7kzYMnNkzQMfwuNOuJqPifBA1P8_BWOoQ@mail.gmail.com> <CAAaarBY4LEC+Wpwg50mrQ7zQQu_po7TQwDKujnq97z2fCdKRew@mail.gmail.com>
From: Dong Lin <lindong28@gmail.com>
Date: Tue, 14 Feb 2017 07:30:07 -0800
Message-ID: <CAAaarBay-no+SE70teQGCbAFxWD9BeDr-JenLR4Ca_yOL+SNcA@mail.gmail.com>
Subject: Re: [DISCUSS] KIP-112: Handle disk failure for JBOD
To: dev@kafka.apache.org
Content-Type: multipart/alternative; boundary=001a114197625f6e3f05487f3e95
archived-at: Tue, 14 Feb 2017 15:30:46 -0000

--001a114197625f6e3f05487f3e95
Content-Type: text/plain; charset=UTF-8

Hey Jun,

I just realized that you may be suggesting that a tool for listing offline
directories is necessary for KIP-112 by asking whether KIP-112 and KIP-113
will be in the same release. I think such a tool is useful but doesn't have
to be included in KIP-112. This is because as of now admin needs to log
into broker machine and check broker log to figure out the cause of broker
failure and the bad log directory in case of disk failure. The KIP-112
won't make it harder since admin can still figure out the bad log directory
by doing the same thing. Thus it is probably OK to just include this script
in KIP-113. Regardless, my hope is to finish both KIPs ASAP and make them
in the same release since both KIPs are needed for the JBOD setup.

Thanks,
Dong

On Mon, Feb 13, 2017 at 5:52 PM, Dong Lin <lindong28@gmail.com> wrote:

> And the test plan has also been updated to simulate disk failure by
> changing log directory permission to 000.
>
> On Mon, Feb 13, 2017 at 5:50 PM, Dong Lin <lindong28@gmail.com> wrote:
>
>> Hi Jun,
>>
>> Thanks for the reply. These comments are very helpful. Let me answer them
>> inline.
>>
>>
>> On Mon, Feb 13, 2017 at 3:25 PM, Jun Rao <jun@confluent.io> wrote:
>>
>>> Hi, Dong,
>>>
>>> Thanks for the reply. A few more replies and new comments below.
>>>
>>> On Fri, Feb 10, 2017 at 4:27 PM, Dong Lin <lindong28@gmail.com> wrote:
>>>
>>> > Hi Jun,
>>> >
>>> > Thanks for the detailed comments. Please see answers inline:
>>> >
>>> > On Fri, Feb 10, 2017 at 3:08 PM, Jun Rao <jun@confluent.io> wrote:
>>> >
>>> > > Hi, Dong,
>>> > >
>>> > > Thanks for the updated wiki. A few comments below.
>>> > >
>>> > > 1. Topics get created
>>> > > 1.1 Instead of storing successfully created replicas in ZK, could we
>>> > store
>>> > > unsuccessfully created replicas in ZK? Since the latter is less
>>> common,
>>> > it
>>> > > probably reduces the load on ZK.
>>> > >
>>> >
>>> > We can store unsuccessfully created replicas in ZK. But I am not sure
>>> if
>>> > that can reduce write load on ZK.
>>> >
>>> > If we want to reduce write load on ZK using by store unsuccessfully
>>> created
>>> > replicas in ZK, then broker should not write to ZK if all replicas are
>>> > successfully created. It means that if /broker/topics/[topic]/partiti
>>> > ons/[partitionId]/controller_managed_state doesn't exist in ZK for a
>>> given
>>> > partition, we have to assume all replicas of this partition have been
>>> > successfully created and send LeaderAndIsrRequest with create = false.
>>> This
>>> > becomes a problem if controller crashes before receiving
>>> > LeaderAndIsrResponse to validate whether a replica has been created.
>>> >
>>> > I think this approach and reduce the number of bytes stored in ZK. But
>>> I am
>>> > not sure if this is a concern.
>>> >
>>> >
>>> >
>>> I was mostly concerned about the controller failover time. Currently, the
>>> controller failover is likely dominated by the cost of reading
>>> topic/partition level information from ZK. If we add another partition
>>> level path in ZK, it probably will double the controller failover time.
>>> If
>>> the approach of representing the non-created replicas doesn't work, have
>>> you considered just adding the created flag in the leaderAndIsr path in
>>> ZK?
>>>
>>>
>> Yes, I have considered adding the created flag in the leaderAndIsr path
>> in ZK. If we were to add created flag per replica in the
>> LeaderAndIsrRequest, then it requires a lot of change in the code base.
>>
>> If we don't add created flag per replica in the LeaderAndIsrRequest, then
>> the information in leaderAndIsr path in ZK and LeaderAndIsrRequest would be
>> different. Further, the procedure for broker to update ISR in ZK will be a
>> bit complicated. When leader updates leaderAndIsr path in ZK, it will have
>> to first read created flags from ZK, change isr, and write leaderAndIsr
>> back to ZK. And it needs to check znode version and re-try write operation
>> in ZK if controller has updated ZK during this period. This is in contrast
>> to the current implementation where the leader either gets all the
>> information from LeaderAndIsrRequest sent by controller, or determine the
>> infromation by itself (e.g. ISR), before writing to leaderAndIsr path in ZK.
>>
>> It seems to me that the above solution is a bit complicated and not
>> clean. Thus I come up with the design in this KIP to store this created
>> flag in a separate zk path. The path is named controller_managed_state to
>> indicate that we can store in this znode all information that is managed by
>> controller only, as opposed to ISR.
>>
>> I agree with your concern of increased ZK read time during controller
>> failover. How about we store the "created" information in the
>> znode /brokers/topics/[topic]? We can change that znode to have the
>> following data format:
>>
>> {
>>   "version" : 2,
>>   "created" : {
>>     "1" : [1, 2, 3],
>>     ...
>>   }
>>   "partition" : {
>>     "1" : [1, 2, 3],
>>     ...
>>   }
>> }
>>
>> We won't have extra zk read using this solution. It also seems reasonable
>> to put the partition assignment information together with replica creation
>> information. The latter is only changed once after the partition is created
>> or re-assigned.
>>
>>
>>>
>>>
>>> >
>>> > > 1.2 If an error is received for a follower, does the controller
>>> eagerly
>>> > > remove it from ISR or do we just let the leader removes it after
>>> timeout?
>>> > >
>>> >
>>> > No, Controller will not actively remove it from ISR. But controller
>>> will
>>> > recognize it as offline replica and propagate this information to all
>>> > brokers via UpdateMetadataRequest. Each leader can use this
>>> information to
>>> > actively remove offline replica from ISR set. I have updated to wiki to
>>> > clarify it.
>>> >
>>> >
>>>
>>> That seems inconsistent with how the controller deals with offline
>>> replicas
>>> due to broker failures. When that happens, the broker will (1) select a
>>> new
>>> leader if the offline replica is the leader; (2) remove the replica from
>>> ISR if the offline replica is the follower. So, intuitively, it seems
>>> that
>>> we should be doing the same thing when dealing with offline replicas due
>>> to
>>> disk failure.
>>>
>>
>> My bad. I misunderstand how the controller currently handles broker
>> failure and ISR change. Yes we should do the same thing when dealing with
>> offline replicas here. I have updated the KIP to specify that, when an
>> offline replica is discovered by controller, the controller removes offline
>> replicas from ISR in the ZK and sends LeaderAndIsrRequest with updated ISR
>> to be used by partition leaders.
>>
>>
>>>
>>>
>>>
>>> >
>>> > > 1.3 Similar, if an error is received for a leader, should the
>>> controller
>>> > > trigger leader election again?
>>> > >
>>> >
>>> > Yes, controller will trigger leader election if leader replica is
>>> offline.
>>> > I have updated the wiki to clarify it.
>>> >
>>> >
>>> > >
>>> > > 2. A log directory stops working on a broker during runtime:
>>> > > 2.1 It seems the broker remembers the failed directory after hitting
>>> an
>>> > > IOException and the failed directory won't be used for creating new
>>> > > partitions until the broker is restarted? If so, could you add that
>>> to
>>> > the
>>> > > wiki.
>>> > >
>>> >
>>> > Right, broker assumes a log directory to be good after it starts, and
>>> mark
>>> > log directory as bad once there is IOException when broker attempts to
>>> > access the log directory. New replicas will only be created on good log
>>> > directory. I just added this to the KIP.
>>> >
>>> >
>>> > > 2.2 Could you be a bit more specific on how and during which
>>> operation
>>> > the
>>> > > broker detects directory failure? Is it when the broker hits an
>>> > IOException
>>> > > during writes, or both reads and writes?  For example, during broker
>>> > > startup, it only reads from each of the log directories, if it hits
>>> an
>>> > > IOException there, does the broker immediately mark the directory as
>>> > > offline?
>>> > >
>>> >
>>> > Broker marks log directory as bad once there is IOException when broker
>>> > attempts to access the log directory. This includes read and write.
>>> These
>>> > operations include log append, log read, log cleaning, watermark
>>> checkpoint
>>> > etc. If broker hits IOException when it reads from each of the log
>>> > directory during startup, it immediately mark the directory as offline.
>>> >
>>> > I just updated the KIP to clarify it.
>>> >
>>> >
>>> > > 3. Partition reassignment: If we know a replica is offline, do we
>>> still
>>> > > want to send StopReplicaRequest to it?
>>> > >
>>> >
>>> > No, controller doesn't send StopReplicaRequest for an offline replica.
>>> > Controller treats this scenario in the same way that exiting Kafka
>>> > implementation does when the broker of this replica is offline.
>>> >
>>> >
>>> > >
>>> > > 4. UpdateMetadataRequestPartitionState: For offline_replicas, do
>>> they
>>> > only
>>> > > include offline replicas due to log directory failures or do they
>>> also
>>> > > include offline replicas due to broker failure?
>>> > >
>>> >
>>> > UpdateMetadataRequestPartitionState's offline_replicas include offline
>>> > replicas due to both log directory failure and broker failure. This is
>>> to
>>> > make the semantics of this field easier to understand. Broker can
>>> > distinguish whether a replica is offline due to broker failure or disk
>>> > failure by checking whether a broker is live in the
>>> UpdateMetadataRequest.
>>> >
>>> >
>>> > >
>>> > > 5. Tools: Could we add some kind of support in the tool to list
>>> offline
>>> > > directories?
>>> > >
>>> >
>>> > In KIP-112 we don't have tools to list offline directories because we
>>> have
>>> > intentionally avoided exposing log directory information (e.g. log
>>> > directory path) to user or other brokers. I think we can add this
>>> feature
>>> > in KIP-113, in which we will have DescribeDirsRequest to list log
>>> directory
>>> > information (e.g. partition assignment, path, size) needed for
>>> rebalance.
>>> >
>>> >
>>> Since we are introducing a new failure mode, if a replica becomes offline
>>> due to failure in log directories, the first thing an admin wants to know
>>> is which log directories are offline from the broker's perspective.  So,
>>> including such a tool will be useful. Do you plan to do KIP-112 and
>>> KIP-113
>>>  in the same release?
>>>
>>>
>> Yes, I agree that including such a tool is using. This is probably better
>> to be added in KIP-113 because we need DescribeDirsRequest to get this
>> information. I will update KIP-113 to include this tool.
>>
>> I plan to do KIP-112 and KIP-113 separately to make each KIP and their
>> patch easier to review. I don't have any plan about which release to have
>> these KIPs. My plan is to both of them ASAP. Is there particular timeline
>> you prefer for code of these two KIPs to checked-in?
>>
>>
>>> >
>>> > >
>>> > > 6. Metrics: Could we add some metrics to show offline directories?
>>> > >
>>> >
>>> > Sure. I think it makes sense to have each broker report its number of
>>> > offline replicas and offline log directories. The previous metric was
>>> put
>>> > in KIP-113. I just added both metrics in KIP-112.
>>> >
>>> >
>>> > >
>>> > > 7. There are still references to kafka-log-dirs.sh. Are they valid?
>>> > >
>>> >
>>> > My bad. I just removed this from "Changes in Operational Procedures"
>>> and
>>> > "Test Plan" in the KIP.
>>> >
>>> >
>>> > >
>>> > > 8. Do you think KIP-113 is ready for review? One thing that KIP-113
>>> > > mentions during partition reassignment is to first send
>>> > > LeaderAndIsrRequest, followed by ChangeReplicaDirRequest. It seems
>>> it's
>>> > > better if the replicas are created in the right log directory in the
>>> > first
>>> > > place? The reason that I brought it up here is because it may affect
>>> the
>>> > > protocol of LeaderAndIsrRequest.
>>> > >
>>> >
>>> > Yes, KIP-113 is ready for review. The advantage of the current design
>>> is
>>> > that we can keep LeaderAndIsrRequest log-direcotry-agnostic. The
>>> > implementation would be much easier to read if all log related logic
>>> (e.g.
>>> > various errors) are put in ChangeReplicadIRrequest and the code path of
>>> > handling replica movement is separated from leadership handling.
>>> >
>>> > In other words, I think Kafka may be easier to develop in the long
>>> term if
>>> > we separate these two requests.
>>> >
>>> > I agree that ideally we want to create replicas in the right log
>>> directory
>>> > in the first place. But I am not sure if there is any performance or
>>> > correctness concern with the existing way of moving it after it is
>>> created.
>>> > Besides, does this decision affect the change proposed in KIP-112?
>>> >
>>> >
>>> I am just wondering if you have considered including the log directory
>>> for
>>> the replicas in the LeaderAndIsrRequest.
>>>
>>>
>> Yeah I have thought about this idea, but only briefly. I rejected this
>> idea because log directory is broker's local information and I prefer not
>> to expose local config information to the cluster through
>> LeaderAndIsrRequest.
>>
>>
>>> 9. Could you describe when the offline replicas due to log directory
>>> failure are removed from the replica fetch threads?
>>>
>>
>> Yes. If the offline replica was a leader, either a new leader is elected
>> or all follower brokers will stop fetching for this partition. If the
>> offline replica is a follower, the broker will stop fetching for this
>> replica immediately. A broker stops fetching data for a replica by removing
>> the replica from the replica fetch threads. I have updated the KIP to
>> clarify it.
>>
>>
>>>
>>> 10. The wiki mentioned changing the log directory to a file for
>>> simulating
>>> disk failure in system tests. Could we just change the permission of the
>>> log directory to 000 to simulate that?
>>>
>>
>>
>> Sure,
>>
>>
>>>
>>> Thanks,
>>>
>>> Jun
>>>
>>>
>>> > > Jun
>>> > >
>>> > > On Fri, Feb 10, 2017 at 9:53 AM, Dong Lin <lindong28@gmail.com>
>>> wrote:
>>> > >
>>> > > > Hi Jun,
>>> > > >
>>> > > > Can I replace zookeeper access with direct RPC for both ISR
>>> > notification
>>> > > > and disk failure notification in a future KIP, or do you feel we
>>> should
>>> > > do
>>> > > > it in this KIP?
>>> > > >
>>> > > > Hi Eno, Grant and everyone,
>>> > > >
>>> > > > Is there further improvement you would like to see with this KIP?
>>> > > >
>>> > > > Thanks you all for the comments,
>>> > > >
>>> > > > Dong
>>> > > >
>>> > > >
>>> > > >
>>> > > > On Thu, Feb 9, 2017 at 4:45 PM, Dong Lin <lindong28@gmail.com>
>>> wrote:
>>> > > >
>>> > > > >
>>> > > > >
>>> > > > > On Thu, Feb 9, 2017 at 3:37 PM, Colin McCabe <cmccabe@apache.org
>>> >
>>> > > wrote:
>>> > > > >
>>> > > > >> On Thu, Feb 9, 2017, at 11:40, Dong Lin wrote:
>>> > > > >> > Thanks for all the comments Colin!
>>> > > > >> >
>>> > > > >> > To answer your questions:
>>> > > > >> > - Yes, a broker will shutdown if all its log directories are
>>> bad.
>>> > > > >>
>>> > > > >> That makes sense.  Can you add this to the writeup?
>>> > > > >>
>>> > > > >
>>> > > > > Sure. This has already been added. You can find it here
>>> > > > > <https://cwiki.apache.org/confluence/pages/diffpagesbyversio
>>> n.action
>>> > ?
>>> > > > pageId=67638402&selectedPageVersions=9&selectedPageVersions=10>
>>> > > > > .
>>> > > > >
>>> > > > >
>>> > > > >>
>>> > > > >> > - I updated the KIP to explicitly state that a log directory
>>> will
>>> > be
>>> > > > >> > assumed to be good until broker sees IOException when it
>>> tries to
>>> > > > access
>>> > > > >> > the log directory.
>>> > > > >>
>>> > > > >> Thanks.
>>> > > > >>
>>> > > > >> > - Controller doesn't explicitly know whether there is new log
>>> > > > directory
>>> > > > >> > or
>>> > > > >> > not. All controller knows is whether replicas are online or
>>> > offline
>>> > > > >> based
>>> > > > >> > on LeaderAndIsrResponse. According to the existing Kafka
>>> > > > implementation,
>>> > > > >> > controller will always send LeaderAndIsrRequest to a broker
>>> after
>>> > it
>>> > > > >> > bounces.
>>> > > > >>
>>> > > > >> I thought so.  It's good to clarify, though.  Do you think it's
>>> > worth
>>> > > > >> adding a quick discussion of this on the wiki?
>>> > > > >>
>>> > > > >
>>> > > > > Personally I don't think it is needed. If broker starts with no
>>> bad
>>> > log
>>> > > > > directory, everything should work it is and we should not need to
>>> > > clarify
>>> > > > > it. The KIP has already covered the scenario when a broker starts
>>> > with
>>> > > > bad
>>> > > > > log directory. Also, the KIP doesn't claim or hint that we
>>> support
>>> > > > dynamic
>>> > > > > addition of new log directories. I think we are good.
>>> > > > >
>>> > > > >
>>> > > > >> best,
>>> > > > >> Colin
>>> > > > >>
>>> > > > >> >
>>> > > > >> > Please see this
>>> > > > >> > <https://cwiki.apache.org/confluence/pages/diffpagesbyversio
>>> > > > >> n.action?pageId=67638402&selectedPageVersions=9&
>>> > > > selectedPageVersions=10>
>>> > > > >> > for the change of the KIP.
>>> > > > >> >
>>> > > > >> > On Thu, Feb 9, 2017 at 11:04 AM, Colin McCabe <
>>> cmccabe@apache.org
>>> > >
>>> > > > >> wrote:
>>> > > > >> >
>>> > > > >> > > On Thu, Feb 9, 2017, at 11:03, Colin McCabe wrote:
>>> > > > >> > > > Thanks, Dong L.
>>> > > > >> > > >
>>> > > > >> > > > Do we plan on bringing down the broker process when all
>>> log
>>> > > > >> directories
>>> > > > >> > > > are offline?
>>> > > > >> > > >
>>> > > > >> > > > Can you explicitly state on the KIP that the log dirs are
>>> all
>>> > > > >> considered
>>> > > > >> > > > good after the broker process is bounced?  It seems like
>>> an
>>> > > > >> important
>>> > > > >> > > > thing to be clear about.  Also, perhaps discuss how the
>>> > > controller
>>> > > > >> > > > becomes aware of the newly good log directories after a
>>> broker
>>> > > > >> bounce
>>> > > > >> > > > (and whether this triggers re-election).
>>> > > > >> > >
>>> > > > >> > > I meant to write, all the log dirs where the broker can
>>> still
>>> > read
>>> > > > the
>>> > > > >> > > index and some other files.  Clearly, log dirs that are
>>> > completely
>>> > > > >> > > inaccessible will still be considered bad after a broker
>>> process
>>> > > > >> bounce.
>>> > > > >> > >
>>> > > > >> > > best,
>>> > > > >> > > Colin
>>> > > > >> > >
>>> > > > >> > > >
>>> > > > >> > > > +1 (non-binding) aside from that
>>> > > > >> > > >
>>> > > > >> > > >
>>> > > > >> > > >
>>> > > > >> > > > On Wed, Feb 8, 2017, at 00:47, Dong Lin wrote:
>>> > > > >> > > > > Hi all,
>>> > > > >> > > > >
>>> > > > >> > > > > Thank you all for the helpful suggestion. I have
>>> updated the
>>> > > KIP
>>> > > > >> to
>>> > > > >> > > > > address
>>> > > > >> > > > > the comments received so far. See here
>>> > > > >> > > > > <https://cwiki.apache.org/conf
>>> > luence/pages/diffpagesbyversio
>>> > > > >> n.action?
>>> > > > >> > > pageId=67638402&selectedPageVe
>>> rsions=8&selectedPageVersions=
>>> > 9>to
>>> > > > >> > > > > read the changes of the KIP. Here is a summary of
>>> change:
>>> > > > >> > > > >
>>> > > > >> > > > > - Updated the Proposed Change section to change the
>>> recovery
>>> > > > >> steps.
>>> > > > >> > > After
>>> > > > >> > > > > this change, broker will also create replica as long as
>>> all
>>> > > log
>>> > > > >> > > > > directories
>>> > > > >> > > > > are working.
>>> > > > >> > > > > - Removed kafka-log-dirs.sh from this KIP since user no
>>> > longer
>>> > > > >> needs to
>>> > > > >> > > > > use
>>> > > > >> > > > > it for recovery from bad disks.
>>> > > > >> > > > > - Explained how the znode controller_managed_state is
>>> > managed
>>> > > in
>>> > > > >> the
>>> > > > >> > > > > Public
>>> > > > >> > > > > interface section.
>>> > > > >> > > > > - Explained what happens during controller failover,
>>> > partition
>>> > > > >> > > > > reassignment
>>> > > > >> > > > > and topic deletion in the Proposed Change section.
>>> > > > >> > > > > - Updated Future Work section to include the following
>>> > > potential
>>> > > > >> > > > > improvements
>>> > > > >> > > > >   - Let broker notify controller of ISR change and disk
>>> > state
>>> > > > >> change
>>> > > > >> > > via
>>> > > > >> > > > > RPC instead of using zookeeper
>>> > > > >> > > > >   - Handle various failure scenarios (e.g. slow disk)
>>> on a
>>> > > > >> case-by-case
>>> > > > >> > > > > basis. For example, we may want to detect slow disk and
>>> > > consider
>>> > > > >> it as
>>> > > > >> > > > > offline.
>>> > > > >> > > > >   - Allow admin to mark a directory as bad so that it
>>> will
>>> > not
>>> > > > be
>>> > > > >> used.
>>> > > > >> > > > >
>>> > > > >> > > > > Thanks,
>>> > > > >> > > > > Dong
>>> > > > >> > > > >
>>> > > > >> > > > >
>>> > > > >> > > > >
>>> > > > >> > > > > On Tue, Feb 7, 2017 at 5:23 PM, Dong Lin <
>>> > lindong28@gmail.com
>>> > > >
>>> > > > >> wrote:
>>> > > > >> > > > >
>>> > > > >> > > > > > Hey Eno,
>>> > > > >> > > > > >
>>> > > > >> > > > > > Thanks much for the comment!
>>> > > > >> > > > > >
>>> > > > >> > > > > > I still think the complexity added to Kafka is
>>> justified
>>> > by
>>> > > > its
>>> > > > >> > > benefit.
>>> > > > >> > > > > > Let me provide my reasons below.
>>> > > > >> > > > > >
>>> > > > >> > > > > > 1) The additional logic is easy to understand and
>>> thus its
>>> > > > >> complexity
>>> > > > >> > > > > > should be reasonable.
>>> > > > >> > > > > >
>>> > > > >> > > > > > On the broker side, it needs to catch exception when
>>> > access
>>> > > > log
>>> > > > >> > > directory,
>>> > > > >> > > > > > mark log directory and all its replicas as offline,
>>> notify
>>> > > > >> > > controller by
>>> > > > >> > > > > > writing the zookeeper notification path, and specify
>>> error
>>> > > in
>>> > > > >> > > > > > LeaderAndIsrResponse. On the controller side, it will
>>> > > listener
>>> > > > >> to
>>> > > > >> > > > > > zookeeper for disk failure notification, learn about
>>> > offline
>>> > > > >> > > replicas in
>>> > > > >> > > > > > the LeaderAndIsrResponse, and take offline replicas
>>> into
>>> > > > >> > > consideration when
>>> > > > >> > > > > > electing leaders. It also mark replica as created in
>>> > > zookeeper
>>> > > > >> and
>>> > > > >> > > use it
>>> > > > >> > > > > > to determine whether a replica is created.
>>> > > > >> > > > > >
>>> > > > >> > > > > > That is all the logic we need to add in Kafka. I
>>> > personally
>>> > > > feel
>>> > > > >> > > this is
>>> > > > >> > > > > > easy to reason about.
>>> > > > >> > > > > >
>>> > > > >> > > > > > 2) The additional code is not much.
>>> > > > >> > > > > >
>>> > > > >> > > > > > I expect the code for KIP-112 to be around 1100 lines
>>> new
>>> > > > code.
>>> > > > >> > > Previously
>>> > > > >> > > > > > I have implemented a prototype of a slightly different
>>> > > design
>>> > > > >> (see
>>> > > > >> > > here
>>> > > > >> > > > > > <https://docs.google.com/docum
>>> ent/d/1Izza0SBmZMVUBUt9s_
>>> > > > >> > > -Dqi3D8e0KGJQYW8xgEdRsgAI/edit>)
>>> > > > >> > > > > > and uploaded it to github (see here
>>> > > > >> > > > > > <https://github.com/lindong28/kafka/tree/JBOD>). The
>>> > patch
>>> > > > >> changed
>>> > > > >> > > 33
>>> > > > >> > > > > > files, added 1185 lines and deleted 183 lines. The
>>> size of
>>> > > > >> prototype
>>> > > > >> > > patch
>>> > > > >> > > > > > is actually smaller than patch of KIP-107 (see here
>>> > > > >> > > > > > <https://github.com/apache/kafka/pull/2476>) which is
>>> > > already
>>> > > > >> > > accepted.
>>> > > > >> > > > > > The KIP-107 patch changed 49 files, added 1349 lines
>>> and
>>> > > > >> deleted 141
>>> > > > >> > > lines.
>>> > > > >> > > > > >
>>> > > > >> > > > > > 3) Comparison with one-broker-per-multiple-volumes
>>> > > > >> > > > > >
>>> > > > >> > > > > > This KIP can improve the availability of Kafka in this
>>> > case
>>> > > > such
>>> > > > >> > > that one
>>> > > > >> > > > > > failed volume doesn't bring down the entire broker.
>>> > > > >> > > > > >
>>> > > > >> > > > > > 4) Comparison with one-broker-per-volume
>>> > > > >> > > > > >
>>> > > > >> > > > > > If each volume maps to multiple disks, then we still
>>> have
>>> > > > >> similar
>>> > > > >> > > problem
>>> > > > >> > > > > > such that the broker will fail if any disk of the
>>> volume
>>> > > > failed.
>>> > > > >> > > > > >
>>> > > > >> > > > > > If each volume maps to one disk, it means that we
>>> need to
>>> > > > >> deploy 10
>>> > > > >> > > > > > brokers on a machine if the machine has 10 disks. I
>>> will
>>> > > > >> explain the
>>> > > > >> > > > > > concern with this approach in order of their
>>> importance.
>>> > > > >> > > > > >
>>> > > > >> > > > > > - It is weird if we were to tell kafka user to deploy
>>> 50
>>> > > > >> brokers on a
>>> > > > >> > > > > > machine of 50 disks.
>>> > > > >> > > > > >
>>> > > > >> > > > > > - Either when user deploys Kafka on a commercial cloud
>>> > > > platform
>>> > > > >> or
>>> > > > >> > > when
>>> > > > >> > > > > > user deploys their own cluster, the size or largest
>>> disk
>>> > is
>>> > > > >> usually
>>> > > > >> > > > > > limited. There will be scenarios where user want to
>>> > increase
>>> > > > >> broker
>>> > > > >> > > > > > capacity by having multiple disks per broker. This
>>> JBOD
>>> > KIP
>>> > > > >> makes it
>>> > > > >> > > > > > feasible without hurting availability due to single
>>> disk
>>> > > > >> failure.
>>> > > > >> > > > > >
>>> > > > >> > > > > > - Automatic load rebalance across disks will be
>>> easier and
>>> > > > more
>>> > > > >> > > flexible
>>> > > > >> > > > > > if one broker has multiple disks. This can be future
>>> work.
>>> > > > >> > > > > >
>>> > > > >> > > > > > - There is performance concern when you deploy 10
>>> broker
>>> > > vs. 1
>>> > > > >> > > broker on
>>> > > > >> > > > > > one machine. The metadata the cluster, including
>>> > > FetchRequest,
>>> > > > >> > > > > > ProduceResponse, MetadataRequest and so on will all
>>> be 10X
>>> > > > >> more. The
>>> > > > >> > > > > > packet-per-second will be 10X higher which may limit
>>> > > > >> performance if
>>> > > > >> > > pps is
>>> > > > >> > > > > > the performance bottleneck. The number of socket on
>>> the
>>> > > > machine
>>> > > > >> is
>>> > > > >> > > 10X
>>> > > > >> > > > > > higher. And the number of replication thread will be
>>> 100X
>>> > > > more.
>>> > > > >> The
>>> > > > >> > > impact
>>> > > > >> > > > > > will be more significant with increasing number of
>>> disks
>>> > per
>>> > > > >> > > machine. Thus
>>> > > > >> > > > > > it will limit Kakfa's scalability in the long term.
>>> > > > >> > > > > >
>>> > > > >> > > > > > Thanks,
>>> > > > >> > > > > > Dong
>>> > > > >> > > > > >
>>> > > > >> > > > > >
>>> > > > >> > > > > > On Tue, Feb 7, 2017 at 1:51 AM, Eno Thereska <
>>> > > > >> eno.thereska@gmail.com
>>> > > > >> > > >
>>> > > > >> > > > > > wrote:
>>> > > > >> > > > > >
>>> > > > >> > > > > >> Hi Dong,
>>> > > > >> > > > > >>
>>> > > > >> > > > > >> To simplify the discussion today, on my part I'll
>>> zoom
>>> > into
>>> > > > one
>>> > > > >> > > thing
>>> > > > >> > > > > >> only:
>>> > > > >> > > > > >>
>>> > > > >> > > > > >> - I'll discuss the options called below :
>>> > > > >> "one-broker-per-disk" or
>>> > > > >> > > > > >> "one-broker-per-few-disks".
>>> > > > >> > > > > >>
>>> > > > >> > > > > >> - I completely buy the JBOD vs RAID arguments so
>>> there is
>>> > > no
>>> > > > >> need to
>>> > > > >> > > > > >> discuss that part for me. I buy it that JBODs are
>>> good.
>>> > > > >> > > > > >>
>>> > > > >> > > > > >> I find the terminology can be improved a bit. Ideally
>>> > we'd
>>> > > be
>>> > > > >> > > talking
>>> > > > >> > > > > >> about volumes, not disks. Just to make it clear that
>>> > Kafka
>>> > > > >> > > understand
>>> > > > >> > > > > >> volumes/directories, not individual raw disks. So by
>>> > > > >> > > > > >> "one-broker-per-few-disks" what I mean is that the
>>> admin
>>> > > can
>>> > > > >> pool a
>>> > > > >> > > few
>>> > > > >> > > > > >> disks together to create a volume/directory and give
>>> that
>>> > > to
>>> > > > >> Kafka.
>>> > > > >> > > > > >>
>>> > > > >> > > > > >>
>>> > > > >> > > > > >> The kernel of my question will be that the admin
>>> already
>>> > > has
>>> > > > >> tools
>>> > > > >> > > to 1)
>>> > > > >> > > > > >> create volumes/directories from a JBOD and 2) start a
>>> > > broker
>>> > > > >> on a
>>> > > > >> > > desired
>>> > > > >> > > > > >> machine and 3) assign a broker resources like a
>>> > directory.
>>> > > I
>>> > > > >> claim
>>> > > > >> > > that
>>> > > > >> > > > > >> those tools are sufficient to optimise resource
>>> > allocation.
>>> > > > I
>>> > > > >> > > understand
>>> > > > >> > > > > >> that a broker could manage point 3) itself, ie
>>> juggle the
>>> > > > >> > > directories. My
>>> > > > >> > > > > >> question is whether the complexity added to Kafka is
>>> > > > justified.
>>> > > > >> > > > > >> Operationally it seems to me an admin will still
>>> have to
>>> > do
>>> > > > >> all the
>>> > > > >> > > three
>>> > > > >> > > > > >> items above.
>>> > > > >> > > > > >>
>>> > > > >> > > > > >> Looking forward to the discussion
>>> > > > >> > > > > >> Thanks
>>> > > > >> > > > > >> Eno
>>> > > > >> > > > > >>
>>> > > > >> > > > > >>
>>> > > > >> > > > > >> > On 1 Feb 2017, at 17:21, Dong Lin <
>>> lindong28@gmail.com
>>> > >
>>> > > > >> wrote:
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > Hey Eno,
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > Thanks much for the review.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > I think your suggestion is to split disks of a
>>> machine
>>> > > into
>>> > > > >> > > multiple
>>> > > > >> > > > > >> disk
>>> > > > >> > > > > >> > sets and run one broker per disk set. Yeah this is
>>> > > similar
>>> > > > to
>>> > > > >> > > Colin's
>>> > > > >> > > > > >> > suggestion of one-broker-per-disk, which we have
>>> > > evaluated
>>> > > > at
>>> > > > >> > > LinkedIn
>>> > > > >> > > > > >> and
>>> > > > >> > > > > >> > considered it to be a good short term approach.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > As of now I don't think any of these approach is a
>>> > better
>>> > > > >> > > alternative in
>>> > > > >> > > > > >> > the long term. I will summarize these here. I have
>>> put
>>> > > > these
>>> > > > >> > > reasons in
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> > KIP's motivation section and rejected alternative
>>> > > section.
>>> > > > I
>>> > > > >> am
>>> > > > >> > > happy to
>>> > > > >> > > > > >> > discuss more and I would certainly like to use an
>>> > > > alternative
>>> > > > >> > > solution
>>> > > > >> > > > > >> that
>>> > > > >> > > > > >> > is easier to do with better performance.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > - JBOD vs. RAID-10: if we switch from RAID-10 with
>>> > > > >> > > > > >> replication-factoer=2 to
>>> > > > >> > > > > >> > JBOD with replicatio-factor=3, we get 25%
>>> reduction in
>>> > > disk
>>> > > > >> usage
>>> > > > >> > > and
>>> > > > >> > > > > >> > doubles the tolerance of broker failure before data
>>> > > > >> > > unavailability from
>>> > > > >> > > > > >> 1
>>> > > > >> > > > > >> > to 2. This is pretty huge gain for any company that
>>> > uses
>>> > > > >> Kafka at
>>> > > > >> > > large
>>> > > > >> > > > > >> > scale.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > - JBOD vs. one-broker-per-disk: The benefit of
>>> > > > >> > > one-broker-per-disk is
>>> > > > >> > > > > >> that
>>> > > > >> > > > > >> > no major code change is needed in Kafka. Among the
>>> > > > >> disadvantage of
>>> > > > >> > > > > >> > one-broker-per-disk summarized in the KIP and
>>> previous
>>> > > > email
>>> > > > >> with
>>> > > > >> > > Colin,
>>> > > > >> > > > > >> > the biggest one is the 15% throughput loss
>>> compared to
>>> > > JBOD
>>> > > > >> and
>>> > > > >> > > less
>>> > > > >> > > > > >> > flexibility to balance across disks. Further, it
>>> > probably
>>> > > > >> requires
>>> > > > >> > > > > >> change
>>> > > > >> > > > > >> > to internal deployment tools at various companies
>>> to
>>> > deal
>>> > > > >> with
>>> > > > >> > > > > >> > one-broker-per-disk setup.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > - JBOD vs. RAID-0: This is the setup that used at
>>> > > > Microsoft.
>>> > > > >> The
>>> > > > >> > > > > >> problem is
>>> > > > >> > > > > >> > that a broker becomes unavailable if any disk fail.
>>> > > Suppose
>>> > > > >> > > > > >> > replication-factor=2 and there are 10 disks per
>>> > machine.
>>> > > > >> Then the
>>> > > > >> > > > > >> > probability of of any message becomes unavailable
>>> due
>>> > to
>>> > > > disk
>>> > > > >> > > failure
>>> > > > >> > > > > >> with
>>> > > > >> > > > > >> > RAID-0 is 100X higher than that with JBOD.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > - JBOD vs. one-broker-per-few-disks:
>>> > > > one-broker-per-few-disk
>>> > > > >> is
>>> > > > >> > > > > >> somewhere
>>> > > > >> > > > > >> > between one-broker-per-disk and RAID-0. So it
>>> carries
>>> > an
>>> > > > >> averaged
>>> > > > >> > > > > >> > disadvantages of these two approaches.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > To answer your question regarding, I think it is
>>> > > reasonable
>>> > > > >> to
>>> > > > >> > > mange
>>> > > > >> > > > > >> disk
>>> > > > >> > > > > >> > in Kafka. By "managing disks" we mean the
>>> management of
>>> > > > >> > > assignment of
>>> > > > >> > > > > >> > replicas across disks. Here are my reasons in more
>>> > > detail:
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > - I don't think this KIP is a big step change. By
>>> > > allowing
>>> > > > >> user to
>>> > > > >> > > > > >> > configure Kafka to run multiple log directories or
>>> > disks
>>> > > as
>>> > > > >> of
>>> > > > >> > > now, it
>>> > > > >> > > > > >> is
>>> > > > >> > > > > >> > implicit that Kafka manages disks. It is just not a
>>> > > > complete
>>> > > > >> > > feature.
>>> > > > >> > > > > >> > Microsoft and probably other companies are using
>>> this
>>> > > > feature
>>> > > > >> > > under the
>>> > > > >> > > > > >> > undesirable effect that a broker will fail any if
>>> any
>>> > > disk
>>> > > > >> fail.
>>> > > > >> > > It is
>>> > > > >> > > > > >> good
>>> > > > >> > > > > >> > to complete this feature.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > - I think it is reasonable to manage disk in
>>> Kafka. One
>>> > > of
>>> > > > >> the
>>> > > > >> > > most
>>> > > > >> > > > > >> > important work that Kafka is doing is to determine
>>> the
>>> > > > >> replica
>>> > > > >> > > > > >> assignment
>>> > > > >> > > > > >> > across brokers and make sure enough copies of a
>>> given
>>> > > > >> replica is
>>> > > > >> > > > > >> available.
>>> > > > >> > > > > >> > I would argue that it is not much different than
>>> > > > determining
>>> > > > >> the
>>> > > > >> > > replica
>>> > > > >> > > > > >> > assignment across disk conceptually.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > - I would agree that this KIP is improve
>>> performance of
>>> > > > >> Kafka at
>>> > > > >> > > the
>>> > > > >> > > > > >> cost
>>> > > > >> > > > > >> > of more complexity inside Kafka, by switching from
>>> > > RAID-10
>>> > > > to
>>> > > > >> > > JBOD. I
>>> > > > >> > > > > >> would
>>> > > > >> > > > > >> > argue that this is a right direction. If we can
>>> gain
>>> > 20%+
>>> > > > >> > > performance by
>>> > > > >> > > > > >> > managing NIC in Kafka as compared to existing
>>> approach
>>> > > and
>>> > > > >> other
>>> > > > >> > > > > >> > alternatives, I would say we should just do it.
>>> Such a
>>> > > gain
>>> > > > >> in
>>> > > > >> > > > > >> performance,
>>> > > > >> > > > > >> > or equivalently reduction in cost, can save
>>> millions of
>>> > > > >> dollars
>>> > > > >> > > per year
>>> > > > >> > > > > >> > for any company running Kafka at large scale.
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > Thanks,
>>> > > > >> > > > > >> > Dong
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> > On Wed, Feb 1, 2017 at 5:41 AM, Eno Thereska <
>>> > > > >> > > eno.thereska@gmail.com>
>>> > > > >> > > > > >> wrote:
>>> > > > >> > > > > >> >
>>> > > > >> > > > > >> >> I'm coming somewhat late to the discussion,
>>> apologies
>>> > > for
>>> > > > >> that.
>>> > > > >> > > > > >> >>
>>> > > > >> > > > > >> >> I'm worried about this proposal. It's moving
>>> Kafka to
>>> > a
>>> > > > >> world
>>> > > > >> > > where it
>>> > > > >> > > > > >> >> manages disks. So in a sense, the scope of the
>>> KIP is
>>> > > > >> limited,
>>> > > > >> > > but the
>>> > > > >> > > > > >> >> direction it sets for Kafka is quite a big step
>>> > change.
>>> > > > >> > > Fundamentally
>>> > > > >> > > > > >> this
>>> > > > >> > > > > >> >> is about balancing resources for a Kafka broker.
>>> This
>>> > > can
>>> > > > be
>>> > > > >> > > done by a
>>> > > > >> > > > > >> >> tool, rather than by changing Kafka. E.g., the
>>> tool
>>> > > would
>>> > > > >> take a
>>> > > > >> > > bunch
>>> > > > >> > > > > >> of
>>> > > > >> > > > > >> >> disks together, create a volume over them and
>>> export
>>> > > that
>>> > > > >> to a
>>> > > > >> > > Kafka
>>> > > > >> > > > > >> broker
>>> > > > >> > > > > >> >> (in addition to setting the memory limits for that
>>> > > broker
>>> > > > or
>>> > > > >> > > limiting
>>> > > > >> > > > > >> other
>>> > > > >> > > > > >> >> resources). A different bunch of disks can then
>>> make
>>> > up
>>> > > a
>>> > > > >> second
>>> > > > >> > > > > >> volume,
>>> > > > >> > > > > >> >> and be used by another Kafka broker. This is
>>> aligned
>>> > > with
>>> > > > >> what
>>> > > > >> > > Colin is
>>> > > > >> > > > > >> >> saying (as I understand it).
>>> > > > >> > > > > >> >>
>>> > > > >> > > > > >> >> Disks are not the only resource on a machine,
>>> there
>>> > are
>>> > > > >> several
>>> > > > >> > > > > >> instances
>>> > > > >> > > > > >> >> where multiple NICs are used for example. Do we
>>> want
>>> > > fine
>>> > > > >> grained
>>> > > > >> > > > > >> >> management of all these resources? I'd argue that
>>> > opens
>>> > > us
>>> > > > >> the
>>> > > > >> > > system
>>> > > > >> > > > > >> to a
>>> > > > >> > > > > >> >> lot of complexity.
>>> > > > >> > > > > >> >>
>>> > > > >> > > > > >> >> Thanks
>>> > > > >> > > > > >> >> Eno
>>> > > > >> > > > > >> >>
>>> > > > >> > > > > >> >>
>>> > > > >> > > > > >> >>> On 1 Feb 2017, at 01:53, Dong Lin <
>>> > lindong28@gmail.com
>>> > > >
>>> > > > >> wrote:
>>> > > > >> > > > > >> >>>
>>> > > > >> > > > > >> >>> Hi all,
>>> > > > >> > > > > >> >>>
>>> > > > >> > > > > >> >>> I am going to initiate the vote If there is no
>>> > further
>>> > > > >> concern
>>> > > > >> > > with
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> >> KIP.
>>> > > > >> > > > > >> >>>
>>> > > > >> > > > > >> >>> Thanks,
>>> > > > >> > > > > >> >>> Dong
>>> > > > >> > > > > >> >>>
>>> > > > >> > > > > >> >>>
>>> > > > >> > > > > >> >>> On Fri, Jan 27, 2017 at 8:08 PM, radai <
>>> > > > >> > > radai.rosenblatt@gmail.com>
>>> > > > >> > > > > >> >> wrote:
>>> > > > >> > > > > >> >>>
>>> > > > >> > > > > >> >>>> a few extra points:
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>>> 1. broker per disk might also incur more client
>>> <-->
>>> > > > >> broker
>>> > > > >> > > sockets:
>>> > > > >> > > > > >> >>>> suppose every producer / consumer "talks" to >1
>>> > > > partition,
>>> > > > >> > > there's a
>>> > > > >> > > > > >> >> very
>>> > > > >> > > > > >> >>>> good chance that partitions that were
>>> co-located on
>>> > a
>>> > > > >> single
>>> > > > >> > > 10-disk
>>> > > > >> > > > > >> >> broker
>>> > > > >> > > > > >> >>>> would now be split between several single-disk
>>> > broker
>>> > > > >> > > processes on
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> >> same
>>> > > > >> > > > > >> >>>> machine. hard to put a multiplier on this, but
>>> > likely
>>> > > > >x1.
>>> > > > >> > > sockets
>>> > > > >> > > > > >> are a
>>> > > > >> > > > > >> >>>> limited resource at the OS level and incur some
>>> > memory
>>> > > > >> cost
>>> > > > >> > > (kernel
>>> > > > >> > > > > >> >>>> buffers)
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>>> 2. there's a memory overhead to spinning up a
>>> JVM
>>> > > > >> (compiled
>>> > > > >> > > code and
>>> > > > >> > > > > >> >> byte
>>> > > > >> > > > > >> >>>> code objects etc). if we assume this overhead is
>>> > ~300
>>> > > MB
>>> > > > >> > > (order of
>>> > > > >> > > > > >> >>>> magnitude, specifics vary) than spinning up 10
>>> JVMs
>>> > > > would
>>> > > > >> lose
>>> > > > >> > > you 3
>>> > > > >> > > > > >> GB
>>> > > > >> > > > > >> >> of
>>> > > > >> > > > > >> >>>> RAM. not a ton, but non negligible.
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>>> 3. there would also be some overhead downstream
>>> of
>>> > > kafka
>>> > > > >> in any
>>> > > > >> > > > > >> >> management
>>> > > > >> > > > > >> >>>> / monitoring / log aggregation system. likely
>>> less
>>> > > than
>>> > > > >> x10
>>> > > > >> > > though.
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>>> 4. (related to above) - added complexity of
>>> > > > administration
>>> > > > >> > > with more
>>> > > > >> > > > > >> >>>> running instances.
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>>> is anyone running kafka with anywhere near 100GB
>>> > > heaps?
>>> > > > i
>>> > > > >> > > thought the
>>> > > > >> > > > > >> >> point
>>> > > > >> > > > > >> >>>> was to rely on kernel page cache to do the disk
>>> > > > buffering
>>> > > > >> ....
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>>> On Thu, Jan 26, 2017 at 11:00 AM, Dong Lin <
>>> > > > >> > > lindong28@gmail.com>
>>> > > > >> > > > > >> wrote:
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>>>> Hey Colin,
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>> Thanks much for the comment. Please see me
>>> comment
>>> > > > >> inline.
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>> On Thu, Jan 26, 2017 at 10:15 AM, Colin McCabe
>>> <
>>> > > > >> > > cmccabe@apache.org>
>>> > > > >> > > > > >> >>>> wrote:
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>> On Wed, Jan 25, 2017, at 13:50, Dong Lin
>>> wrote:
>>> > > > >> > > > > >> >>>>>>> Hey Colin,
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> Good point! Yeah we have actually considered
>>> and
>>> > > > >> tested this
>>> > > > >> > > > > >> >>>> solution,
>>> > > > >> > > > > >> >>>>>>> which we call one-broker-per-disk. It would
>>> work
>>> > > and
>>> > > > >> should
>>> > > > >> > > > > >> require
>>> > > > >> > > > > >> >>>> no
>>> > > > >> > > > > >> >>>>>>> major change in Kafka as compared to this
>>> JBOD
>>> > KIP.
>>> > > > So
>>> > > > >> it
>>> > > > >> > > would
>>> > > > >> > > > > >> be a
>>> > > > >> > > > > >> >>>>> good
>>> > > > >> > > > > >> >>>>>>> short term solution.
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> But it has a few drawbacks which makes it
>>> less
>>> > > > >> desirable in
>>> > > > >> > > the
>>> > > > >> > > > > >> long
>>> > > > >> > > > > >> >>>>>>> term.
>>> > > > >> > > > > >> >>>>>>> Assume we have 10 disks on a machine. Here
>>> are
>>> > the
>>> > > > >> problems:
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>> Hi Dong,
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>> Thanks for the thoughtful reply.
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> 1) Our stress test result shows that
>>> > > > >> one-broker-per-disk
>>> > > > >> > > has 15%
>>> > > > >> > > > > >> >>>> lower
>>> > > > >> > > > > >> >>>>>>> throughput
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> 2) Controller would need to send 10X as many
>>> > > > >> > > LeaderAndIsrRequest,
>>> > > > >> > > > > >> >>>>>>> MetadataUpdateRequest and StopReplicaRequest.
>>> > This
>>> > > > >> > > increases the
>>> > > > >> > > > > >> >>>> burden
>>> > > > >> > > > > >> >>>>>>> on
>>> > > > >> > > > > >> >>>>>>> controller which can be the performance
>>> > bottleneck.
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>> Maybe I'm misunderstanding something, but
>>> there
>>> > > would
>>> > > > >> not be
>>> > > > >> > > 10x as
>>> > > > >> > > > > >> >>>> many
>>> > > > >> > > > > >> >>>>>> StopReplicaRequest RPCs, would there?  The
>>> other
>>> > > > >> requests
>>> > > > >> > > would
>>> > > > >> > > > > >> >>>> increase
>>> > > > >> > > > > >> >>>>>> 10x, but from a pretty low base, right?  We
>>> are
>>> > not
>>> > > > >> > > reassigning
>>> > > > >> > > > > >> >>>>>> partitions all the time, I hope (or else we
>>> have
>>> > > > bigger
>>> > > > >> > > > > >> problems...)
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>> I think the controller will group
>>> > StopReplicaRequest
>>> > > > per
>>> > > > >> > > broker and
>>> > > > >> > > > > >> >> send
>>> > > > >> > > > > >> >>>>> only one StopReplicaRequest to a broker during
>>> > > > controlled
>>> > > > >> > > shutdown.
>>> > > > >> > > > > >> >>>> Anyway,
>>> > > > >> > > > > >> >>>>> we don't have to worry about this if we agree
>>> that
>>> > > > other
>>> > > > >> > > requests
>>> > > > >> > > > > >> will
>>> > > > >> > > > > >> >>>>> increase by 10X. One MetadataRequest to send to
>>> > each
>>> > > > >> broker
>>> > > > >> > > in the
>>> > > > >> > > > > >> >>>> cluster
>>> > > > >> > > > > >> >>>>> every time there is leadership change. I am not
>>> > sure
>>> > > > >> this is
>>> > > > >> > > a real
>>> > > > >> > > > > >> >>>>> problem. But in theory this makes the overhead
>>> > > > complexity
>>> > > > >> > > O(number
>>> > > > >> > > > > >> of
>>> > > > >> > > > > >> >>>>> broker) and may be a concern in the future.
>>> Ideally
>>> > > we
>>> > > > >> should
>>> > > > >> > > avoid
>>> > > > >> > > > > >> it.
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> 3) Less efficient use of physical resource
>>> on the
>>> > > > >> machine.
>>> > > > >> > > The
>>> > > > >> > > > > >> number
>>> > > > >> > > > > >> >>>>> of
>>> > > > >> > > > > >> >>>>>>> socket on each machine will increase by 10X.
>>> The
>>> > > > >> number of
>>> > > > >> > > > > >> connection
>>> > > > >> > > > > >> >>>>>>> between any two machine will increase by
>>> 100X.
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> 4) Less efficient way to management memory
>>> and
>>> > > quota.
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> 5) Rebalance between disks/brokers on the
>>> same
>>> > > > machine
>>> > > > >> will
>>> > > > >> > > less
>>> > > > >> > > > > >> >>>>>>> efficient
>>> > > > >> > > > > >> >>>>>>> and less flexible. Broker has to read data
>>> from
>>> > > > another
>>> > > > >> > > broker on
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> >>>>>>> same
>>> > > > >> > > > > >> >>>>>>> machine via socket. It is also harder to do
>>> > > automatic
>>> > > > >> load
>>> > > > >> > > balance
>>> > > > >> > > > > >> >>>>>>> between
>>> > > > >> > > > > >> >>>>>>> disks on the same machine in the future.
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> I will put this and the explanation in the
>>> > rejected
>>> > > > >> > > alternative
>>> > > > >> > > > > >> >>>>> section.
>>> > > > >> > > > > >> >>>>>>> I
>>> > > > >> > > > > >> >>>>>>> have a few questions:
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> - Can you explain why this solution can help
>>> > avoid
>>> > > > >> > > scalability
>>> > > > >> > > > > >> >>>>>>> bottleneck?
>>> > > > >> > > > > >> >>>>>>> I actually think it will exacerbate the
>>> > scalability
>>> > > > >> problem
>>> > > > >> > > due
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> >>>> 2)
>>> > > > >> > > > > >> >>>>>>> above.
>>> > > > >> > > > > >> >>>>>>> - Why can we push more RPC with this
>>> solution?
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>> To really answer this question we'd have to
>>> take a
>>> > > > deep
>>> > > > >> dive
>>> > > > >> > > into
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> >>>>>> locking of the broker and figure out how
>>> > effectively
>>> > > > it
>>> > > > >> can
>>> > > > >> > > > > >> >> parallelize
>>> > > > >> > > > > >> >>>>>> truly independent requests.  Almost every
>>> > > > multithreaded
>>> > > > >> > > process is
>>> > > > >> > > > > >> >>>> going
>>> > > > >> > > > > >> >>>>>> to have shared state, like shared queues or
>>> shared
>>> > > > >> sockets,
>>> > > > >> > > that is
>>> > > > >> > > > > >> >>>>>> going to make scaling less than linear when
>>> you
>>> > add
>>> > > > >> disks or
>>> > > > >> > > > > >> >>>> processors.
>>> > > > >> > > > > >> >>>>>> (And clearly, another option is to improve
>>> that
>>> > > > >> scalability,
>>> > > > >> > > rather
>>> > > > >> > > > > >> >>>>>> than going multi-process!)
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>> Yeah I also think it is better to improve
>>> > scalability
>>> > > > >> inside
>>> > > > >> > > kafka
>>> > > > >> > > > > >> code
>>> > > > >> > > > > >> >>>> if
>>> > > > >> > > > > >> >>>>> possible. I am not sure we currently have any
>>> > > > scalability
>>> > > > >> > > issue
>>> > > > >> > > > > >> inside
>>> > > > >> > > > > >> >>>>> Kafka that can not be removed without using
>>> > > > >> multi-process.
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>>> - It is true that a garbage collection in one
>>> > > broker
>>> > > > >> would
>>> > > > >> > > not
>>> > > > >> > > > > >> affect
>>> > > > >> > > > > >> >>>>>>> others. But that is after every broker only
>>> uses
>>> > > 1/10
>>> > > > >> of the
>>> > > > >> > > > > >> memory.
>>> > > > >> > > > > >> >>>>> Can
>>> > > > >> > > > > >> >>>>>>> we be sure that this will actually help
>>> > > performance?
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>> The big question is, how much memory do Kafka
>>> > > brokers
>>> > > > >> use
>>> > > > >> > > now, and
>>> > > > >> > > > > >> how
>>> > > > >> > > > > >> >>>>>> much will they use in the future?  Our
>>> experience
>>> > in
>>> > > > >> HDFS
>>> > > > >> > > was that
>>> > > > >> > > > > >> >> once
>>> > > > >> > > > > >> >>>>>> you start getting more than 100-200GB Java
>>> heap
>>> > > sizes,
>>> > > > >> full
>>> > > > >> > > GCs
>>> > > > >> > > > > >> start
>>> > > > >> > > > > >> >>>>>> taking minutes to finish when using the
>>> standard
>>> > > JVMs.
>>> > > > >> That
>>> > > > >> > > alone
>>> > > > >> > > > > >> is
>>> > > > >> > > > > >> >> a
>>> > > > >> > > > > >> >>>>>> good reason to go multi-process or consider
>>> > storing
>>> > > > more
>>> > > > >> > > things off
>>> > > > >> > > > > >> >> the
>>> > > > >> > > > > >> >>>>>> Java heap.
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>> I see. Now I agree one-broker-per-disk should
>>> be
>>> > more
>>> > > > >> > > efficient in
>>> > > > >> > > > > >> >> terms
>>> > > > >> > > > > >> >>>> of
>>> > > > >> > > > > >> >>>>> GC since each broker probably needs less than
>>> 1/10
>>> > of
>>> > > > the
>>> > > > >> > > memory
>>> > > > >> > > > > >> >>>> available
>>> > > > >> > > > > >> >>>>> on a typical machine nowadays. I will remove
>>> this
>>> > > from
>>> > > > >> the
>>> > > > >> > > reason of
>>> > > > >> > > > > >> >>>>> rejection.
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>> Disk failure is the "easy" case.  The "hard"
>>> case,
>>> > > > >> which is
>>> > > > >> > > > > >> >>>>>> unfortunately also the much more common case,
>>> is
>>> > > disk
>>> > > > >> > > misbehavior.
>>> > > > >> > > > > >> >>>>>> Towards the end of their lives, disks tend to
>>> > start
>>> > > > >> slowing
>>> > > > >> > > down
>>> > > > >> > > > > >> >>>>>> unpredictably.  Requests that would have
>>> completed
>>> > > > >> > > immediately
>>> > > > >> > > > > >> before
>>> > > > >> > > > > >> >>>>>> start taking 20, 100 500 milliseconds.  Some
>>> files
>>> > > may
>>> > > > >> be
>>> > > > >> > > readable
>>> > > > >> > > > > >> and
>>> > > > >> > > > > >> >>>>>> other files may not be.  System calls hang,
>>> > > sometimes
>>> > > > >> > > forever, and
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> >>>>>> Java process can't abort them, because the
>>> hang is
>>> > > in
>>> > > > >> the
>>> > > > >> > > kernel.
>>> > > > >> > > > > >> It
>>> > > > >> > > > > >> >>>> is
>>> > > > >> > > > > >> >>>>>> not fun when threads are stuck in "D state"
>>> > > > >> > > > > >> >>>>>> http://stackoverflow.com/quest
>>> > > > >> ions/20423521/process-perminan
>>> > > > >> > > > > >> >>>>>> tly-stuck-on-d-state
>>> > > > >> > > > > >> >>>>>> .  Even kill -9 cannot abort the thread then.
>>> > > > >> Fortunately,
>>> > > > >> > > this is
>>> > > > >> > > > > >> >>>>>> rare.
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>> I agree it is a harder problem and it is rare.
>>> We
>>> > > > >> probably
>>> > > > >> > > don't
>>> > > > >> > > > > >> have
>>> > > > >> > > > > >> >> to
>>> > > > >> > > > > >> >>>>> worry about it in this KIP since this issue is
>>> > > > >> orthogonal to
>>> > > > >> > > > > >> whether or
>>> > > > >> > > > > >> >>>> not
>>> > > > >> > > > > >> >>>>> we use JBOD.
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>> Another approach we should consider is for
>>> Kafka
>>> > to
>>> > > > >> > > implement its
>>> > > > >> > > > > >> own
>>> > > > >> > > > > >> >>>>>> storage layer that would stripe across
>>> multiple
>>> > > disks.
>>> > > > >> This
>>> > > > >> > > > > >> wouldn't
>>> > > > >> > > > > >> >>>>>> have to be done at the block level, but could
>>> be
>>> > > done
>>> > > > >> at the
>>> > > > >> > > file
>>> > > > >> > > > > >> >>>> level.
>>> > > > >> > > > > >> >>>>>> We could use consistent hashing to determine
>>> which
>>> > > > >> disks a
>>> > > > >> > > file
>>> > > > >> > > > > >> should
>>> > > > >> > > > > >> >>>>>> end up on, for example.
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>> Are you suggesting that we should distribute
>>> log,
>>> > or
>>> > > > log
>>> > > > >> > > segment,
>>> > > > >> > > > > >> >> across
>>> > > > >> > > > > >> >>>>> disks of brokers? I am not sure if I fully
>>> > understand
>>> > > > >> this
>>> > > > >> > > > > >> approach. My
>>> > > > >> > > > > >> >>>> gut
>>> > > > >> > > > > >> >>>>> feel is that this would be a drastic solution
>>> that
>>> > > > would
>>> > > > >> > > require
>>> > > > >> > > > > >> >>>>> non-trivial design. While this may be useful to
>>> > > Kafka,
>>> > > > I
>>> > > > >> would
>>> > > > >> > > > > >> prefer
>>> > > > >> > > > > >> >> not
>>> > > > >> > > > > >> >>>>> to discuss this in detail in this thread
>>> unless you
>>> > > > >> believe
>>> > > > >> > > it is
>>> > > > >> > > > > >> >>>> strictly
>>> > > > >> > > > > >> >>>>> superior to the design in this KIP in terms of
>>> > > solving
>>> > > > >> our
>>> > > > >> > > use-case.
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>>> best,
>>> > > > >> > > > > >> >>>>>> Colin
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> Thanks,
>>> > > > >> > > > > >> >>>>>>> Dong
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>> On Wed, Jan 25, 2017 at 11:34 AM, Colin
>>> McCabe <
>>> > > > >> > > > > >> cmccabe@apache.org>
>>> > > > >> > > > > >> >>>>>>> wrote:
>>> > > > >> > > > > >> >>>>>>>
>>> > > > >> > > > > >> >>>>>>>> Hi Dong,
>>> > > > >> > > > > >> >>>>>>>>
>>> > > > >> > > > > >> >>>>>>>> Thanks for the writeup!  It's very
>>> interesting.
>>> > > > >> > > > > >> >>>>>>>>
>>> > > > >> > > > > >> >>>>>>>> I apologize in advance if this has been
>>> > discussed
>>> > > > >> > > somewhere else.
>>> > > > >> > > > > >> >>>>> But
>>> > > > >> > > > > >> >>>>>> I
>>> > > > >> > > > > >> >>>>>>>> am curious if you have considered the
>>> solution
>>> > of
>>> > > > >> running
>>> > > > >> > > > > >> multiple
>>> > > > >> > > > > >> >>>>>>>> brokers per node.  Clearly there is a memory
>>> > > > overhead
>>> > > > >> with
>>> > > > >> > > this
>>> > > > >> > > > > >> >>>>>> solution
>>> > > > >> > > > > >> >>>>>>>> because of the fixed cost of starting
>>> multiple
>>> > > JVMs.
>>> > > > >> > > However,
>>> > > > >> > > > > >> >>>>> running
>>> > > > >> > > > > >> >>>>>>>> multiple JVMs would help avoid scalability
>>> > > > >> bottlenecks.
>>> > > > >> > > You
>>> > > > >> > > > > >> could
>>> > > > >> > > > > >> >>>>>>>> probably push more RPCs per second, for
>>> example.
>>> > > A
>>> > > > >> garbage
>>> > > > >> > > > > >> >>>>> collection
>>> > > > >> > > > > >> >>>>>>>> in one broker would not affect the others.
>>> It
>>> > > would
>>> > > > >> be
>>> > > > >> > > > > >> interesting
>>> > > > >> > > > > >> >>>>> to
>>> > > > >> > > > > >> >>>>>>>> see this considered in the "alternate
>>> designs"
>>> > > > design,
>>> > > > >> > > even if
>>> > > > >> > > > > >> you
>>> > > > >> > > > > >> >>>>> end
>>> > > > >> > > > > >> >>>>>>>> up deciding it's not the way to go.
>>> > > > >> > > > > >> >>>>>>>>
>>> > > > >> > > > > >> >>>>>>>> best,
>>> > > > >> > > > > >> >>>>>>>> Colin
>>> > > > >> > > > > >> >>>>>>>>
>>> > > > >> > > > > >> >>>>>>>>
>>> > > > >> > > > > >> >>>>>>>> On Thu, Jan 12, 2017, at 10:46, Dong Lin
>>> wrote:
>>> > > > >> > > > > >> >>>>>>>>> Hi all,
>>> > > > >> > > > > >> >>>>>>>>>
>>> > > > >> > > > > >> >>>>>>>>> We created KIP-112: Handle disk failure for
>>> > JBOD.
>>> > > > >> Please
>>> > > > >> > > find
>>> > > > >> > > > > >> the
>>> > > > >> > > > > >> >>>>> KIP
>>> > > > >> > > > > >> >>>>>>>>> wiki
>>> > > > >> > > > > >> >>>>>>>>> in the link https://cwiki.apache.org/confl
>>> > > > >> > > > > >> >>>> uence/display/KAFKA/KIP-
>>> > > > >> > > > > >> >>>>>>>>> 112%3A+Handle+disk+failure+for+JBOD.
>>> > > > >> > > > > >> >>>>>>>>>
>>> > > > >> > > > > >> >>>>>>>>> This KIP is related to KIP-113
>>> > > > >> > > > > >> >>>>>>>>> <https://cwiki.apache.org/conf
>>> > > > >> luence/display/KAFKA/KIP-
>>> > > > >> > > > > >> >>>>>>>> 113%3A+Support+replicas+moveme
>>> > > > >> nt+between+log+directories>:
>>> > > > >> > > > > >> >>>>>>>>> Support replicas movement between log
>>> > > directories.
>>> > > > >> They
>>> > > > >> > > are
>>> > > > >> > > > > >> >>>> needed
>>> > > > >> > > > > >> >>>>> in
>>> > > > >> > > > > >> >>>>>>>>> order
>>> > > > >> > > > > >> >>>>>>>>> to support JBOD in Kafka. Please help
>>> review
>>> > the
>>> > > > >> KIP. You
>>> > > > >> > > > > >> >>>> feedback
>>> > > > >> > > > > >> >>>>> is
>>> > > > >> > > > > >> >>>>>>>>> appreciated!
>>> > > > >> > > > > >> >>>>>>>>>
>>> > > > >> > > > > >> >>>>>>>>> Thanks,
>>> > > > >> > > > > >> >>>>>>>>> Dong
>>> > > > >> > > > > >> >>>>>>>>
>>> > > > >> > > > > >> >>>>>>
>>> > > > >> > > > > >> >>>>>
>>> > > > >> > > > > >> >>>>
>>> > > > >> > > > > >> >>
>>> > > > >> > > > > >> >>
>>> > > > >> > > > > >>
>>> > > > >> > > > > >>
>>> > > > >> > > > > >
>>> > > > >> > >
>>> > > > >>
>>> > > > >
>>> > > > >
>>> > > >
>>> > >
>>> >
>>>
>>
>>
>

--001a114197625f6e3f05487f3e95--