Hi Jun, Please look at the earlier reply and let us know your comments. Thanks, Satish.
On Wed, Feb 12, 2020 at 4:06 PM Satish Duggana <satish.dugg...@gmail.com> wrote: > > Hi Jun, > Thanks for your comments on the separation of remote log metadata > storage and remote log storage. > We had a few discussions since early Jan on how to support eventually > consistent stores like S3 by uncoupling remote log segment metadata > and remote log storage. It is written with details in the doc here(1). > Below is the brief summary of the discussion from that doc. > > The current approach consists of pulling the remote log segment > metadata from remote log storage APIs. It worked fine for storages > like HDFS. But one of the problems of relying on the remote storage to > maintain metadata is that tiered-storage needs to be strongly > consistent, with an impact not only on the metadata(e.g. LIST in S3) > but also on the segment data(e.g. GET after a DELETE in S3). The cost > of maintaining metadata in remote storage needs to be factored in. > This is true in the case of S3, LIST APIs incur huge costs as you > raised earlier. > So, it is good to separate the remote storage from the remote log > metadata store. We refactored the existing RemoteStorageManager and > introduced RemoteLogMetadataManager. Remote log metadata store should > give strong consistency semantics but remote log storage can be > eventually consistent. > We can have a default implementation for RemoteLogMetadataManager > which uses an internal topic(as mentioned in one of our earlier > emails) as storage. But users can always plugin their own > RemoteLogMetadataManager implementation based on their environment. > > Please go through the updated KIP and let us know your comments. We > have started refactoring for the changes mentioned in the KIP and > there may be a few more updates to the APIs. > > [1] > https://docs.google.com/document/d/1qfkBCWL1e7ZWkHU7brxKDBebq4ie9yK20XJnKbgAlew/edit?ts=5e208ec7# > > On Fri, Dec 27, 2019 at 5:43 PM Ivan Yurchenko <ivan0yurche...@gmail.com> > wrote: > > > > Hi all, > > > > > > Jun: > > > (a) Cost: S3 list object requests cost $0.005 per 1000 requests. If you > > > have 100,000 partitions and want to pull the metadata for each partition > > at > > > the rate of 1/sec. It can cost $0.5/sec, which is roughly $40K per day. > > > > I want to note here, that no reasonably durable storage will be cheap > > at 100k RPS. For example, DynamoDB might give the same ballpark figures. > > If we want to keep the pull-based approach, we can try to reduce this number > > in several ways: doing listings less frequently (as Satish mentioned, > > with the current defaults it's ~3.33k RPS for your example), > > batching listing operations in some way (depending on the storage; > > it might require the change of RSM's interface). > > > > > > > There are different ways for doing push based metadata propagation. Some > > > object stores may support that already. For example, S3 supports events > > > notification > > This sounds interesting. However, I see a couple of issues using it: > > 1. As I understand the documentation, notification delivery is not > > guaranteed > > and it's recommended to periodically do LIST to fill the gaps. > > Which brings us back to the same LIST consistency guarantees issue. > > 2. The same goes for the broker start: to get the current state, we need > > to LIST. > > 3. The dynamic set of multiple consumers (RSMs): AFAIK SQS and SNS aren't > > designed for such a case. > > > > > > Alexandre: > > > A.1 As commented on PR 7561, S3 consistency model [1][2] implies RSM > > cannot > > > relies solely on S3 APIs to guarantee the expected strong consistency. The > > > proposed implementation [3] would need to be updated to take this into > > > account. Let’s talk more about this. > > > > Thank you for the feedback. I clearly see the need for changing the S3 > > implementation > > to provide stronger consistency guarantees. As it see from this thread, > > there are > > several possible approaches to this. Let's discuss RemoteLogManager's > > contract and > > behavior (like pull vs push model) further before picking one (or several - > > ?) of them. > > I'm going to do some evaluation of DynamoDB for the pull-based approach, > > if it's possible to apply it paying a reasonable bill. Also, of the > > push-based approach > > with a Kafka topic as the medium. > > > > > > > A.2.3 Atomicity – what does an implementation of RSM need to provide with > > > respect to atomicity of the APIs copyLogSegment, cleanupLogUntil and > > > deleteTopicPartition? If a partial failure happens in any of those (e.g. > > in > > > the S3 implementation, if one of the multiple uploads fails [4]), > > > > The S3 implementation is going to change, but it's worth clarifying anyway. > > The segment log file is being uploaded after S3 has acked uploading of > > all other files associated with the segment and only after this the whole > > segment file set becomes visible remotely for operations like > > listRemoteSegments [1]. > > In case of upload failure, the files that has been successfully uploaded > > stays > > as invisible garbage that is collected by cleanupLogUntil (or overwritten > > successfully later). > > And the opposite happens during the deletion: log files are deleted first. > > This approach should generally work when we solve consistency issues > > by adding a strongly consistent storage: a segment's uploaded files remain > > invisible garbage until some metadata about them is written. > > > > > > > A.3 Caching – storing locally the segments retrieved from the remote > > > storage is excluded as it does not align with the original intent and even > > > defeat some of its purposes (save disk space etc.). That said, could there > > > be other types of use cases where the pattern of access to the remotely > > > stored segments would benefit from local caching (and potentially > > > read-ahead)? Consider the use case of a large pool of consumers which > > start > > > a backfill at the same time for one day worth of data from one year ago > > > stored remotely. Caching the segments locally would allow to uncouple the > > > load on the remote storage from the load on the Kafka cluster. Maybe the > > > RLM could expose a configuration parameter to switch that feature on/off? > > > > I tend to agree here, caching remote segments locally and making > > this configurable sounds pretty practical to me. We should implement this, > > maybe not in the first iteration. > > > > > > Br, > > Ivan > > > > [1] > > https://github.com/harshach/kafka/pull/18/files#diff-4d73d01c16caed6f2548fc3063550ef0R152 > > > > On Thu, 19 Dec 2019 at 19:49, Alexandre Dupriez > > <alexandre.dupr...@gmail.com> > > wrote: > > > > > Hi Jun, > > > > > > Thank you for the feedback. I am trying to understand how a push-based > > > approach would work. > > > In order for the metadata to be propagated (under the assumption you > > > stated), would you plan to add a new API in Kafka to allow the > > > metadata store to send them directly to the brokers? > > > > > > Thanks, > > > Alexandre > > > > > > > > > Le mer. 18 déc. 2019 à 20:14, Jun Rao <j...@confluent.io> a écrit : > > > > > > > > Hi, Satish and Ivan, > > > > > > > > There are different ways for doing push based metadata propagation. Some > > > > object stores may support that already. For example, S3 supports events > > > > notification ( > > > > https://docs.aws.amazon.com/AmazonS3/latest/dev/NotificationHowTo.html). > > > > Otherwise one could use a separate metadata store that supports > > > push-based > > > > change propagation. Other people have mentioned using a Kafka topic. The > > > > best approach may depend on the object store and the operational > > > > environment (e.g. whether an external metadata store is already > > > available). > > > > > > > > The above discussion is based on the assumption that we need to cache > > > > the > > > > object metadata locally in every broker. I mentioned earlier that an > > > > alternative is to just store/retrieve those metadata in an external > > > > metadata store. That may simplify the implementation in some cases. > > > > > > > > Thanks, > > > > > > > > Jun > > > > > > > > On Thu, Dec 5, 2019 at 7:01 AM Satish Duggana <satish.dugg...@gmail.com> > > > > wrote: > > > > > > > > > Hi Jun, > > > > > Thanks for your reply. > > > > > > > > > > Currently, `listRemoteSegments` is called at the configured > > > > > interval(not every second, defaults to 30secs). Storing remote log > > > > > metadata in a strongly consistent store for S3 RSM is raised in > > > > > PR-comment[1]. > > > > > RLM invokes RSM at regular intervals and RSM can give remote segment > > > > > metadata if it is available. RSM is responsible for maintaining and > > > > > fetching those entries. It should be based on whatever mechanism is > > > > > consistent and efficient with the respective remote storage. > > > > > > > > > > Can you give more details about push based mechanism from RSM? > > > > > > > > > > 1. https://github.com/apache/kafka/pull/7561#discussion_r344576223 > > > > > > > > > > Thanks, > > > > > Satish. > > > > > > > > > > On Thu, Dec 5, 2019 at 4:23 AM Jun Rao <j...@confluent.io> wrote: > > > > > > > > > > > > Hi, Harsha, > > > > > > > > > > > > Thanks for the reply. > > > > > > > > > > > > 40/41. I am curious which block storages you have tested. S3 seems > > > to be > > > > > > one of the popular block stores. The concerns that I have with pull > > > based > > > > > > approach are the following. > > > > > > (a) Cost: S3 list object requests cost $0.005 per 1000 requests. If > > > you > > > > > > have 100,000 partitions and want to pull the metadata for each > > > partition > > > > > at > > > > > > the rate of 1/sec. It can cost $0.5/sec, which is roughly $40K per > > > day. > > > > > > (b) Semantics: S3 list objects are eventually consistent. So, when > > > you > > > > > do a > > > > > > list object request, there is no guarantee that you can see all > > > uploaded > > > > > > objects. This could impact the correctness of subsequent logics. > > > > > > (c) Efficiency: Blindly pulling metadata when there is no change > > > > > > adds > > > > > > unnecessary overhead in the broker as well as in the block store. > > > > > > > > > > > > So, have you guys tested S3? If so, could you share your experience > > > in > > > > > > terms of cost, semantics and efficiency? > > > > > > > > > > > > Jun > > > > > > > > > > > > > > > > > > On Tue, Dec 3, 2019 at 10:11 PM Harsha Chintalapani <ka...@harsha.io > > > > > > > > > wrote: > > > > > > > > > > > > > Hi Jun, > > > > > > > Thanks for the reply. > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Nov 26, 2019 at 3:46 PM, Jun Rao <j...@confluent.io> > > > > > > > wrote: > > > > > > > > > > > > > > > Hi, Satish and Ying, > > > > > > > > > > > > > > > > Thanks for the reply. > > > > > > > > > > > > > > > > 40/41. There are two different ways that we can approach this. > > > One is > > > > > > > what > > > > > > > > you said. We can have an opinionated way of storing and > > > populating > > > > > the > > > > > > > tier > > > > > > > > metadata that we think is good enough for everyone. I am not > > > sure if > > > > > this > > > > > > > > is the case based on what's currently proposed in the KIP. For > > > > > example, I > > > > > > > > am not sure that (1) everyone always needs local metadata; (2) > > > the > > > > > > > current > > > > > > > > local storage format is general enough and (3) everyone wants to > > > use > > > > > the > > > > > > > > pull based approach to propagate the metadata. Another approach > > > is to > > > > > > > make > > > > > > > > this pluggable and let the implementor implements the best > > > approach > > > > > for a > > > > > > > > particular block storage. I haven't seen any comments from > > > > > Slack/AirBnb > > > > > > > in > > > > > > > > the mailing list on this topic. It would be great if they can > > > provide > > > > > > > > feedback directly here. > > > > > > > > > > > > > > > > > > > > > > The current interfaces are designed with most popular block > > > storages > > > > > > > available today and we did 2 implementations with these > > > interfaces and > > > > > > > they both are yielding good results as we going through the > > > testing of > > > > > it. > > > > > > > If there is ever a need for pull based approach we can definitely > > > > > evolve > > > > > > > the interface. > > > > > > > In the past we did mark interfaces to be evolving to make room for > > > > > unknowns > > > > > > > in the future. > > > > > > > If you have any suggestions around the current interfaces please > > > > > propose we > > > > > > > are happy to see if we can work them into it. > > > > > > > > > > > > > > > > > > > > > 43. To offer tier storage as a general feature, ideally all > > > existing > > > > > > > > capabilities should still be supported. It's fine if the uber > > > > > > > > implementation doesn't support all capabilities for internal > > > usage. > > > > > > > > However, the framework should be general enough. > > > > > > > > > > > > > > > > > > > > > > We agree on that as a principle. But all of these major features > > > mostly > > > > > > > coming right now and to have a new big feature such as tiered > > > storage > > > > > to > > > > > > > support all the new features will be a big ask. We can document on > > > how > > > > > do > > > > > > > we approach solving these in future iterations. > > > > > > > Our goal is to make this tiered storage feature work for everyone. > > > > > > > > > > > > > > 43.3 This is more than just serving the tier-ed data from block > > > > > storage. > > > > > > > > With KIP-392, the consumer now can resolve the conflicts with > > > > > > > > the > > > > > replica > > > > > > > > based on leader epoch. So, we need to make sure that leader > > > > > > > > epoch > > > > > can be > > > > > > > > recovered properly from tier storage. > > > > > > > > > > > > > > > > > > > > > > We are working on testing our approach and we will update the KIP > > > with > > > > > > > design details. > > > > > > > > > > > > > > 43.4 For JBOD, if tier storage stores the tier metadata locally, > > > > > > > we > > > > > need to > > > > > > > > support moving such metadata across disk directories since JBOD > > > > > supports > > > > > > > > moving data across disks. > > > > > > > > > > > > > > > > > > > > > > KIP is updated with JBOD details. Having said that JBOD tooling > > > needs > > > > > to > > > > > > > evolve to support production loads. Most of the users will be > > > > > interested in > > > > > > > using tiered storage without JBOD support support on day 1. > > > > > > > > > > > > > > Thanks, > > > > > > > Harsha > > > > > > > > > > > > > > As for meeting, we could have a KIP e-meeting on this if needed, > > > but it > > > > > > > > will be open to everyone and will be recorded and shared. Often, > > > the > > > > > > > > details are still resolved through the mailing list. > > > > > > > > > > > > > > > > Jun > > > > > > > > > > > > > > > > On Tue, Nov 19, 2019 at 6:48 PM Ying Zheng > > > <yi...@uber.com.invalid> > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > Please ignore my previous email > > > > > > > > I didn't know Apache requires all the discussions to be "open" > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Nov 19, 2019, 5:40 PM Ying Zheng <yi...@uber.com> wrote: > > > > > > > > > > > > > > > > Hi Jun, > > > > > > > > > > > > > > > > Thank you very much for your feedback! > > > > > > > > > > > > > > > > Can we schedule a meeting in your Palo Alto office in December? > > > > > > > > I > > > > > think a > > > > > > > > face to face discussion is much more efficient than emails. Both > > > > > Harsha > > > > > > > > > > > > > > > > and > > > > > > > > > > > > > > > > I can visit you. Satish may be able to join us remotely. > > > > > > > > > > > > > > > > On Fri, Nov 15, 2019 at 11:04 AM Jun Rao <j...@confluent.io> > > > wrote: > > > > > > > > > > > > > > > > Hi, Satish and Harsha, > > > > > > > > > > > > > > > > The following is a more detailed high level feedback for the > > > > > > > > KIP. > > > > > > > > > > > > > > > > Overall, > > > > > > > > > > > > > > > > the KIP seems useful. The challenge is how to design it such > > > > > > > > that > > > > > it’s > > > > > > > > general enough to support different ways of implementing this > > > feature > > > > > > > > > > > > > > > > and > > > > > > > > > > > > > > > > support existing features. > > > > > > > > > > > > > > > > 40. Local segment metadata storage: The KIP makes the assumption > > > that > > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > > metadata for the archived log segments are cached locally in > > > every > > > > > > > > > > > > > > > > broker > > > > > > > > > > > > > > > > and provides a specific implementation for the local storage in > > > the > > > > > > > > framework. We probably should discuss this more. For example, > > > some > > > > > tier > > > > > > > > storage providers may not want to cache the metadata locally and > > > just > > > > > > > > > > > > > > > > rely > > > > > > > > > > > > > > > > > > > > > > > > upon a remote key/value store if such a store is already > > > present. If > > > > > a > > > > > > > > local store is used, there could be different ways of > > > implementing it > > > > > > > > (e.g., based on customized local files, an embedded local store > > > like > > > > > > > > RocksDB, etc). An alternative of designing this is to just > > > provide an > > > > > > > > interface for retrieving the tier segment metadata and leave the > > > > > details > > > > > > > of > > > > > > > > how to get the metadata outside of the framework. > > > > > > > > > > > > > > > > > > > > > > > > 41. RemoteStorageManager interface and the usage of the > > > interface in > > > > > the > > > > > > > > framework: I am not sure if the interface is general enough. For > > > > > > > > > > > > > > > > example, > > > > > > > > > > > > > > > > it seems that RemoteLogIndexEntry is tied to a specific way of > > > > > storing > > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > > metadata in remote storage. The framework uses > > > listRemoteSegments() > > > > > api > > > > > > > > > > > > > > > > in > > > > > > > > > > > > > > > > > > > > > > > > a pull based approach. However, in some other implementations, a > > > push > > > > > > > > based > > > > > > > > approach may be more preferred. I don’t have a concrete proposal > > > yet. > > > > > > > > > > > > > > > > > > > > > > > > But, > > > > > > > > > > > > > > > > it would be useful to give this area some more thoughts and see > > > if we > > > > > > > > > > > > > > > > can > > > > > > > > > > > > > > > > make the interface more general. > > > > > > > > > > > > > > > > 42. In the diagram, the RemoteLogManager is side by side with > > > > > > > > > > > > > > > > LogManager. > > > > > > > > > > > > > > > > This KIP only discussed how the fetch request is handled between > > > the > > > > > two > > > > > > > > layer. However, we should also consider how other requests that > > > touch > > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > > log can be handled. e.g., list offsets by timestamp, delete > > > records, > > > > > > > > > > > > > > > > etc. > > > > > > > > > > > > > > > > Also, in this model, it's not clear which component is > > > responsible > > > > > for > > > > > > > > managing the log start offset. It seems that the log start > > > > > > > > offset > > > > > could > > > > > > > > > > > > > > > > be > > > > > > > > > > > > > > > > changed by both RemoteLogManager and LogManager. > > > > > > > > > > > > > > > > > > > > > > > > 43. There are quite a few existing features not covered by the > > > KIP. > > > > > It > > > > > > > > would be useful to discuss each of those. > > > > > > > > 43.1 I won’t say that compacted topics are rarely used and > > > > > > > > always > > > > > small. > > > > > > > > For example, KStreams uses compacted topics for storing the > > > states > > > > > and > > > > > > > > sometimes the size of the topic could be large. While it might > > > be ok > > > > > to > > > > > > > not > > > > > > > > support compacted topics initially, it would be useful to have a > > > high > > > > > > > > level > > > > > > > > idea on how this might be supported down the road so that we > > > don’t > > > > > have > > > > > > > > > > > > > > > > > > > > > > > > to > > > > > > > > > > > > > > > > > > > > > > > > make incompatible API changes in the future. > > > > > > > > 43.2 We need to discuss how EOS is supported. In particular, how > > > is > > > > > the > > > > > > > > producer state integrated with the remote storage. 43.3 Now that > > > > > KIP-392 > > > > > > > > (allow consumers to fetch from closest replica) is implemented, > > > we > > > > > need > > > > > > > to > > > > > > > > discuss how reading from a follower replica is supported with > > > tier > > > > > > > storage. > > > > > > > > 43.4 We need to discuss how JBOD is supported with tier storage. > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > > > > Jun > > > > > > > > > > > > > > > > On Fri, Nov 8, 2019 at 12:06 AM Tom Bentley <tbent...@redhat.com > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > Thanks for those insights Ying. > > > > > > > > > > > > > > > > On Thu, Nov 7, 2019 at 9:26 PM Ying Zheng > > > > > > > > <yi...@uber.com.invalid > > > > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > Thanks, I missed that point. However, there's still a point at > > > > > > > > > > > > > > > > which > > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > > consumer fetches start getting served from remote storage (even > > > if > > > > > > > > > > > > > > > > that > > > > > > > > > > > > > > > > point isn't as soon as the local log retention time/size). This > > > > > > > > > > > > > > > > represents > > > > > > > > > > > > > > > > a kind of performance cliff edge and what I'm really interested > > > in > > > > > > > > > > > > > > > > is > > > > > > > > > > > > > > > > how > > > > > > > > > > > > > > > > easy it is for a consumer which falls off that cliff to catch up > > > > > > > > > > > > > > > > and so > > > > > > > > > > > > > > > > its > > > > > > > > > > > > > > > > fetches again come from local storage. Obviously this can depend > > > > > > > > > > > > > > > > on > > > > > > > > > > > > > > > > all > > > > > > > > > > > > > > > > sorts of factors (like production rate, consumption rate), so > > > it's > > > > > > > > > > > > > > > > not > > > > > > > > > > > > > > > > guaranteed (just like it's not guaranteed for Kafka today), but > > > > > > > > > > > > > > > > this > > > > > > > > > > > > > > > > would > > > > > > > > > > > > > > > > represent a new failure mode. > > > > > > > > > > > > > > > > > > > > > > > > As I have explained in the last mail, it's a very rare case that > > > a > > > > > > > > consumer > > > > > > > > need to read remote data. With our experience at Uber, this only > > > > > > > > > > > > > > > > > > > > > > > > happens > > > > > > > > > > > > > > > > when the consumer service had an outage for several hours. > > > > > > > > > > > > > > > > There is not a "performance cliff" as you assume. The remote > > > storage > > > > > > > > > > > > > > > > is > > > > > > > > > > > > > > > > even faster than local disks in terms of bandwidth. Reading from > > > > > > > > > > > > > > > > remote > > > > > > > > > > > > > > > > storage is going to have higher latency than local disk. But > > > since > > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > > > > > > > > > > consumer > > > > > > > > is catching up several hours data, it's not sensitive to the > > > > > > > > > > > > > > > > > > > > > > > > sub-second > > > > > > > > > > > > > > > > > > > > > > > > level > > > > > > > > latency, and each remote read request will read a large amount > > > > > > > > of > > > > > > > > > > > > > > > > > > > > > > > > data to > > > > > > > > > > > > > > > > make the overall performance better than reading from local > > > disks. > > > > > > > > > > > > > > > > Another aspect I'd like to understand better is the effect of > > > > > > > > > > > > > > > > serving > > > > > > > > > > > > > > > > fetch > > > > > > > > > > > > > > > > request from remote storage has on the broker's network > > > > > > > > > > > > > > > > utilization. If > > > > > > > > > > > > > > > > we're just trimming the amount of data held locally (without > > > > > > > > > > > > > > > > increasing > > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > > overall local+remote retention), then we're effectively trading > > > > > > > > > > > > > > > > disk > > > > > > > > > > > > > > > > bandwidth for network bandwidth when serving fetch requests from > > > > > > > > > > > > > > > > remote > > > > > > > > > > > > > > > > storage (which I understand to be a good thing, since brokers > > > > > > > > are > > > > > > > > often/usually disk bound). But if we're increasing the overall > > > > > > > > > > > > > > > > local+remote > > > > > > > > > > > > > > > > retention then it's more likely that network itself becomes the > > > > > > > > > > > > > > > > bottleneck. > > > > > > > > > > > > > > > > I appreciate this is all rather hand wavy, I'm just trying to > > > > > > > > > > > > > > > > understand > > > > > > > > > > > > > > > > how this would affect broker performance, so I'd be grateful for > > > > > > > > > > > > > > > > any > > > > > > > > > > > > > > > > insights you can offer. > > > > > > > > > > > > > > > > Network bandwidth is a function of produce speed, it has nothing > > > to > > > > > > > > > > > > > > > > do > > > > > > > > > > > > > > > > with > > > > > > > > > > > > > > > > remote retention. As long as the data is shipped to remote > > > storage, > > > > > > > > > > > > > > > > you > > > > > > > > > > > > > > > > can > > > > > > > > > > > > > > > > keep the data there for 1 day or 1 year or 100 years, it doesn't > > > > > > > > > > > > > > > > consume > > > > > > > > > > > > > > > > > > > > > > > > any > > > > > > > > network resources. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
