Thanks for patiently addressing all the comments. I will add the vote in the other thread.
-- Divij Vaidya On Mon, Nov 4, 2024 at 5:40 AM Kamal Chandraprakash < kamal.chandraprak...@gmail.com> wrote: > Hi all, > > If the review comments are addressed and KIP looks good, please submit your > vote on the voting thread. > > Thanks, > Kamal > > On Sat, Nov 2, 2024 at 7:52 PM Kamal Chandraprakash < > kamal.chandraprak...@gmail.com> wrote: > > > I've opened a PR which adds an integration test > > < > https://github.com/apache/kafka/pull/17668/files#diff-a60b518846fc0f770164d55b6d7cd31e03a002514377578c80c6e22cc120af40R88 > > > > to show the impact of this change/KIP. PTAL. > > > > -- > > Kamal > > > > On Fri, Nov 1, 2024 at 3:31 PM Kamal Chandraprakash < > > kamal.chandraprak...@gmail.com> wrote: > > > >> Thanks for the review! > >> > >> > 5 > >> Updated the KIP to include the `isEmpty` method in the transaction index > >> > >> > 6 > >> You're right. The offset parameter will be equal to the > >> next-segment-to-consider base offset. > >> But, the API introduced in RLMM is for *one* epoch. The next epoch > >> start-offset may not be > >> the base-offset. You can refer to this line > >> < > https://github.com/apache/kafka/pull/17659/files#diff-380e4d8859ea9148f21794c09039425c82d9012a392c2dbbe1ce2ec8677a1970R1857 > > > >> in the draft PR. > >> > >> > >> On Fri, Nov 1, 2024 at 3:16 PM Divij Vaidya <divijvaidy...@gmail.com> > >> wrote: > >> > >>> Last few things - > >>> > >>> # 5 > >>> About setting the TrxIndexEmpty field, could we introduce an isEmpty() > >>> function in TransactionIndex which has the following implementation: > >>> > >>> public boolean isEmpty() { > >>> return !iterable().iterator().hasNext(); > >>> } > >>> > >>> > >>> The advantages of this approach is: > >>> 1. It works for both cases when the file is not present and also when > the > >>> file is present but is empty. > >>> 2. It prevents leaking the underlying implementation of > TransactionIndex > >>> outside via the file() method. I think that making file() as public is > an > >>> implementation leak (for example, what is the trx indx is not file > >>> based!). > >>> > >>> > >>> #6 > >>> In the documentation for nextSegmentWithTxnIndex, the offset parameter > >>> should be equal to the next-segment-to-consider's base offset, no? > >>> I assume that we will add a new fetch here with nextSegmentBaseOffset > >>> > >>> > https://github.com/apache/kafka/blob/346fdbafc539bc48bb66eedae89a15e240007fd9/core/src/main/java/kafka/log/remote/RemoteLogManager.java#L1801 > >>> . Is there a case where the parameter "offset" will not be equal to the > >>> baseOffset of a segment? > >>> > >>> -- > >>> Divij Vaidya > >>> > >>> > >>> > >>> On Fri, Nov 1, 2024 at 10:26 AM Kamal Chandraprakash < > >>> kamal.chandraprak...@gmail.com> wrote: > >>> > >>> > Hi Divij, > >>> > > >>> > Thanks for the detailed review! > >>> > > >>> > > 1, 2, 3, 4 > >>> > Updated the KIP-1058 > >>> > < > >>> > > >>> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-1058%3A+Txn+consumer+exerts+pressure+on+remote+storage+when+collecting+aborted+transactions > >>> > > > >>> > with the feedback received and also opened a draft PR for #17659 > >>> > <https://github.com/apache/kafka/pull/17659/files> reference. > >>> > PTAL. > >>> > > >>> > > 5. How are we determining the value of the TrxIndexEmpty field on > >>> segment > >>> > rotation? > >>> > Transaction index file is optional, the file does not exists when > >>> there are > >>> > no aborted txn entries for a > >>> > segment, we will be using the file null check. Also, updated it in > the > >>> KIP. > >>> > > >>> > Thanks, > >>> > Kamal > >>> > > >>> > > >>> > On Tue, Oct 29, 2024 at 8:47 PM Divij Vaidya < > divijvaidy...@gmail.com> > >>> > wrote: > >>> > > >>> > > A few more points to discuss (please add to the KIP as well) > >>> > > > >>> > > 5. How are we determining the value of the TrxIndexEmpty field on > >>> segment > >>> > > rotation? > >>> > > > >>> > > One option is to do a boolean txnIdxEmpty = > >>> > > segment.txnIndex().allAbortedTxns().isEmpty() but this will have an > >>> > > overhead of reading the contents of the file and storing them in > >>> memory, > >>> > > when we have a non-empty index. > >>> > > The other option (preferred) is to add an isEmpty() public method > to > >>> the > >>> > > TransactionIndex and perform a segment.txnIndex().isEmpty() check > >>> which > >>> > > will internally use Files.size() java API. > >>> > > > >>> > > On Tue, Oct 29, 2024 at 1:21 PM Divij Vaidya < > >>> divijvaidy...@gmail.com> > >>> > > wrote: > >>> > > > >>> > > > Let's get the ball rolling (again) on this one. > >>> > > > > >>> > > > Kamal, could you please add the following to the KIP: > >>> > > > 1. the API as discussed above. Please add the failure modes for > >>> this > >>> > API > >>> > > > as well such as the exceptions thrown and a recommendation on > how a > >>> > > caller > >>> > > > is expected to handle those. I am assuming that the three > >>> parameters > >>> > for > >>> > > > this API will be topicPartition, epoch and offset. > >>> > > > 2. implementation details for Topic based RLMM. I am assuming > that > >>> the > >>> > > > plugin will default the field to false if this field is absent > >>> (case of > >>> > > old > >>> > > > metadata). > >>> > > > 3. In the test plan section, additionally, we need to assert that > >>> we > >>> > > don't > >>> > > > read metadata for all segments (i.e. it is not a linear search) > >>> from > >>> > the > >>> > > > Topic based RLMM. > >>> > > > 4. in the compatibility section, please document how the existing > >>> > > clusters > >>> > > > with Tiered Storage metadata will work during/after a rolling > >>> upgrade > >>> > to > >>> > > a > >>> > > > version which contains this new change. > >>> > > > > >>> > > > -- > >>> > > > Divij Vaidya > >>> > > > > >>> > > > > >>> > > > > >>> > > > On Fri, Oct 11, 2024 at 12:26 PM Kamal Chandraprakash < > >>> > > > kamal.chandraprak...@gmail.com> wrote: > >>> > > > > >>> > > >> Bump for review. > >>> > > >> > >>> > > >> If the additional proposal looks good, I'll append them to the > >>> KIP. > >>> > > PTAL. > >>> > > >> > >>> > > >> New API in RLMM#nextRemoteLogSegmentMetadataWithTxnIndex > >>> > > >> > >>> > > >> -- > >>> > > >> Kamal > >>> > > >> > >>> > > >> On Sun, Oct 6, 2024 at 7:20 PM Kamal Chandraprakash < > >>> > > >> kamal.chandraprak...@gmail.com> wrote: > >>> > > >> > >>> > > >> > Hi Christo, > >>> > > >> > > >>> > > >> > Thanks for the review! > >>> > > >> > > >>> > > >> > Adding the new API `nextRemoteLogSegmentMetadataWithTxnIndex` > in > >>> > RLMM > >>> > > >> > helps to > >>> > > >> > reduce the complexity of linear search. With this API, we have > >>> to: > >>> > > >> > > >>> > > >> > 1. Maintain one more skip-list [1] for each of the epochs in > the > >>> > > >> partition > >>> > > >> > in RLMM that might > >>> > > >> > increase the memory usage of TopicBased RLMM > implementation. > >>> > > >> > 1a) The skip-list will be empty when there are no aborted > >>> txn > >>> > > >> entries > >>> > > >> > for a partition/epoch which is the predominant case. > >>> > > >> > 1b) The skip-list will act as a duplicate when *most* of > the > >>> > > >> segments > >>> > > >> > have aborted txn entries, assuming aborted txn are quite low, > >>> this > >>> > > >> should > >>> > > >> > be fine. > >>> > > >> > 2. Change the logic to retrieve the aborted txns (we have to > >>> query > >>> > the > >>> > > >> > nextRLSMWithTxnIndex > >>> > > >> > for each of the leader-epoch). > >>> > > >> > 3. Logic divergence from how we retrieve the aborted txn > entries > >>> > > >> compared > >>> > > >> > to the local-log. > >>> > > >> > > >>> > > >> > The approach looks good to me. If everyone is aligned, then we > >>> can > >>> > > >> proceed > >>> > > >> > to add this API to RLMM. > >>> > > >> > > >>> > > >> > Another option I was thinking of is to capture the > >>> > > `lastStableOffsetLag` > >>> > > >> > [2] while rotating the segment. > >>> > > >> > But, that is a bigger change we can take later. > >>> > > >> > > >>> > > >> > [1]: > >>> > > >> > > >>> > > >> > >>> > > > >>> > > >>> > https://sourcegraph.com/github.com/apache/kafka/-/blob/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogLeaderEpochState.java?L43 > >>> > > >> > [2]: > >>> > > >> > > >>> > > >> > >>> > > > >>> > > >>> > https://sourcegraph.com/github.com/apache/kafka/-/blob/core/src/main/scala/kafka/log/UnifiedLog.scala?L432 > >>> > > >> > > >>> > > >> > > >>> > > >> > Thanks, > >>> > > >> > Kamal > >>> > > >> > > >>> > > >> > On Fri, Oct 4, 2024 at 4:21 PM Christo Lolov < > >>> > christolo...@gmail.com> > >>> > > >> > wrote: > >>> > > >> > > >>> > > >> >> Heya, > >>> > > >> >> > >>> > > >> >> Apologies for the delay. I have been thinking about this > >>> problem > >>> > > >> recently > >>> > > >> >> as well and while I believe storing a boolean in the metadata > >>> is > >>> > > good, > >>> > > >> I > >>> > > >> >> think we can do better by introducing a new method to the > RLMM > >>> > along > >>> > > >> the > >>> > > >> >> lines of > >>> > > >> >> > >>> > > >> >> Optional<RemoteLogSegmentMetadata> > >>> > > >> >> nextRemoteLogSegmentMetadataWithTxnIndex(TopicIdPartition > >>> > > >> >> topicIdPartition, > >>> > > >> >> int epochForOffset, long offset) throws > RemoteStorageException > >>> > > >> >> > >>> > > >> >> This will help plugin implementers to build optimisations > such > >>> as > >>> > > skip > >>> > > >> >> lists which will give them the next segment quicker than a > >>> linear > >>> > > >> search. > >>> > > >> >> > >>> > > >> >> I am keen to hear your thoughts! > >>> > > >> >> > >>> > > >> >> Best, > >>> > > >> >> Christo > >>> > > >> >> > >>> > > >> >> On Fri, 4 Oct 2024 at 10:48, Kamal Chandraprakash < > >>> > > >> >> kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > >>> > > >> >> > Hi Luke, > >>> > > >> >> > > >>> > > >> >> > Thanks for the review! > >>> > > >> >> > > >>> > > >> >> > > Do you think it is helpful if we store the "least abort > >>> start > >>> > > >> offset > >>> > > >> >> in > >>> > > >> >> > the > >>> > > >> >> > segment", and -1 means no txnIndex. So that we can have a > >>> way to > >>> > > know > >>> > > >> >> if we > >>> > > >> >> > need to fetch this txn index or not. > >>> > > >> >> > > >>> > > >> >> > 1. No, this change won't have an effect. To find the > >>> upper-bound > >>> > > >> offset > >>> > > >> >> > [1], we have to > >>> > > >> >> > fetch that segment's offset index file. The > >>> RemoteIndexCache > >>> > > [2] > >>> > > >> >> > fetches all the 3 > >>> > > >> >> > index files together and caches them for subsequent > use, > >>> so > >>> > > this > >>> > > >> >> > improvement > >>> > > >> >> > won't have an effect as the current segment txn index > >>> gets > >>> > > >> >> downloaded > >>> > > >> >> > anyway. > >>> > > >> >> > > >>> > > >> >> > 2. The reason for choosing boolean is to make the change > >>> backward > >>> > > >> >> > compatible. > >>> > > >> >> > There can be existing RLM events for the uploaded > >>> segments. > >>> > > The > >>> > > >> >> > default > >>> > > >> >> > value of `txnIdxEmpty` is false so the *old* RLM > events > >>> are > >>> > > >> >> assumed to > >>> > > >> >> > contain > >>> > > >> >> > the txn index files and those files are downloaded if > >>> they > >>> > > >> exist. > >>> > > >> >> > > >>> > > >> >> > [1]: > >>> > > >> >> > > >>> > > >> >> > > >>> > > >> >> > >>> > > >> > >>> > > > >>> > > >>> > https://sourcegraph.com/github.com/apache/kafka@trunk/-/blob/core/src/main/java/kafka/log/remote/RemoteLogManager.java?L1732 > >>> > > >> >> > [2]: > >>> > > >> >> > > >>> > > >> >> > > >>> > > >> >> > >>> > > >> > >>> > > > >>> > > >>> > https://sourcegraph.com/github.com/apache/kafka@trunk/-/blob/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java?L383 > >>> > > >> >> > > >>> > > >> >> > Thanks, > >>> > > >> >> > Kamal > >>> > > >> >> > > >>> > > >> >> > On Thu, Oct 3, 2024 at 3:11 PM Luke Chen < > show...@gmail.com> > >>> > > wrote: > >>> > > >> >> > > >>> > > >> >> > > Hi Kamal, > >>> > > >> >> > > > >>> > > >> >> > > Sorry for the late review. > >>> > > >> >> > > Thanks for the KIP, this will improve the transaction > >>> reading > >>> > for > >>> > > >> >> remote > >>> > > >> >> > > storage. > >>> > > >> >> > > Overall LGTM, just one minor thought: > >>> > > >> >> > > > >>> > > >> >> > > Currently, we only store the `TxnIndexEmpty` bool value > in > >>> the > >>> > > >> segment > >>> > > >> >> > > metadata. > >>> > > >> >> > > Do you think it is helpful if we store the "least abort > >>> start > >>> > > >> offset > >>> > > >> >> in > >>> > > >> >> > the > >>> > > >> >> > > segment", and -1 means no txnIndex. So that we can have a > >>> way > >>> > to > >>> > > >> know > >>> > > >> >> if > >>> > > >> >> > we > >>> > > >> >> > > need to fetch this txn index or not. > >>> > > >> >> > > > >>> > > >> >> > > Thanks. > >>> > > >> >> > > Luke > >>> > > >> >> > > > >>> > > >> >> > > On Mon, Sep 9, 2024 at 3:26 PM Kamal Chandraprakash < > >>> > > >> >> > > kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > > > >>> > > >> >> > > > Hi all, > >>> > > >> >> > > > > >>> > > >> >> > > > If there are no more comments, I'll start a voting > thread > >>> > soon. > >>> > > >> >> > > > > >>> > > >> >> > > > Thanks, > >>> > > >> >> > > > Kamal > >>> > > >> >> > > > > >>> > > >> >> > > > On Fri, Sep 6, 2024 at 7:28 PM Kamal Chandraprakash < > >>> > > >> >> > > > kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > > > > >>> > > >> >> > > > > Bumping this thread again for review! > >>> > > >> >> > > > > > >>> > > >> >> > > > > Reduced the scope of the proposal to minimum. We will > >>> be > >>> > > adding > >>> > > >> >> only > >>> > > >> >> > > one > >>> > > >> >> > > > > field (txnIdxEmpty) to the > >>> > > >> >> > > > > RemoteLogSegmentMetadata event which is backward > >>> > compatible. > >>> > > >> PTAL. > >>> > > >> >> > > > > > >>> > > >> >> > > > > Thanks, > >>> > > >> >> > > > > Kamal > >>> > > >> >> > > > > > >>> > > >> >> > > > > > >>> > > >> >> > > > > On Tue, Aug 13, 2024 at 6:33 PM Kamal Chandraprakash > < > >>> > > >> >> > > > > kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > > > > > >>> > > >> >> > > > >> Bumping this thread for KIP review! > >>> > > >> >> > > > >> > >>> > > >> >> > > > >> We can go for the simplest solution that is proposed > >>> in > >>> > this > >>> > > >> KIP > >>> > > >> >> and > >>> > > >> >> > > > >> it can be improved in the subsequent iteration. > PTAL. > >>> > > >> >> > > > >> > >>> > > >> >> > > > >> Thanks, > >>> > > >> >> > > > >> Kamal > >>> > > >> >> > > > >> > >>> > > >> >> > > > >> On Fri, Aug 2, 2024 at 11:42 AM Kamal > Chandraprakash < > >>> > > >> >> > > > >> kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > > > >> > >>> > > >> >> > > > >>> Hi Divij, > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> Thanks for the review! And, sorry for the late > reply. > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> From the UnifiedLog.scala > >>> > > >> >> > > > >>> < > >>> > > >> >> > > > > >>> > > >> >> > > > >>> > > >> >> > > >>> > > >> >> > >>> > > >> > >>> > > > >>> > > >>> > https://sourcegraph.com/github.com/apache/kafka@trunk/-/blob/core/src/main/scala/kafka/log/UnifiedLog.scala?L421-427 > >>> > > >> >> > > > > > >>> > > >> >> > > > >>> doc: > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> """ > >>> > > >> >> > > > >>> The last stable offset (LSO) is defined as the > first > >>> > offset > >>> > > >> such > >>> > > >> >> > that > >>> > > >> >> > > > >>> all lower offsets have been "decided." > >>> > > >> >> > > > >>> * Non-transactional messages are considered > >>> decided > >>> > > >> >> immediately, > >>> > > >> >> > > but > >>> > > >> >> > > > >>> transactional messages are only decided when > >>> > > >> >> > > > >>> * the corresponding COMMIT or ABORT marker is > >>> written. > >>> > > >> This > >>> > > >> >> > > implies > >>> > > >> >> > > > >>> that the last stable offset will be equal > >>> > > >> >> > > > >>> * to the high watermark if there are no > >>> transactional > >>> > > >> >> messages > >>> > > >> >> > in > >>> > > >> >> > > > the > >>> > > >> >> > > > >>> log. Note also that the LSO cannot advance > >>> > > >> >> > > > >>> * beyond the high watermark. > >>> > > >> >> > > > >>> """ > >>> > > >> >> > > > >>> While rolling the active segment to passive, if LSO > >>> > equals > >>> > > to > >>> > > >> >> HW, > >>> > > >> >> > > then > >>> > > >> >> > > > >>> all the messages in that segment are > >>> > > >> >> > > > >>> decided and we can store the `lastStableOffsetLag` > >>> as an > >>> > > >> >> attribute > >>> > > >> >> > in > >>> > > >> >> > > > >>> the rolled segment. We can then propagate > >>> > > >> >> > > > >>> the `lastStableOffsetLag` information in the > >>> > > >> RemoteLogMetadata > >>> > > >> >> > > events. > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> While reading the remote log segment, if the > >>> > > >> >> `lastStableOffsetLag` > >>> > > >> >> > is > >>> > > >> >> > > > 0, > >>> > > >> >> > > > >>> then there is no need to traverse to > >>> > > >> >> > > > >>> the subsequent segments for aborted transactions > >>> which > >>> > > covers > >>> > > >> >> the > >>> > > >> >> > > case > >>> > > >> >> > > > >>> for the dominant case where the > >>> > > >> >> > > > >>> partition had no transactions at all. > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> With Log compaction, the shrinked segments might > get > >>> > > merged. > >>> > > >> One > >>> > > >> >> > > option > >>> > > >> >> > > > >>> is to take the max of `lastStableOffsetLag` > >>> > > >> >> > > > >>> and store it in the new LogSegment. Since, the > tiered > >>> > > storage > >>> > > >> >> does > >>> > > >> >> > > not > >>> > > >> >> > > > >>> support compacted topics / historical compacted > >>> > > >> >> > > > >>> topics, we can omit this case. > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> If this approach looks good, I can update the KIP > >>> with > >>> > the > >>> > > >> >> details. > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> -- > >>> > > >> >> > > > >>> Kamal > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>> On Tue, Jun 25, 2024 at 4:24 PM Divij Vaidya < > >>> > > >> >> > > divijvaidy...@gmail.com> > >>> > > >> >> > > > >>> wrote: > >>> > > >> >> > > > >>> > >>> > > >> >> > > > >>>> Hi Kamal > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> Thanks for the bump. I have been thinking about > this > >>> > > >> passively > >>> > > >> >> for > >>> > > >> >> > > the > >>> > > >> >> > > > >>>> past > >>> > > >> >> > > > >>>> few days. > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> The simplest solution is to store a state at > segment > >>> > level > >>> > > >> >> > metadata. > >>> > > >> >> > > > The > >>> > > >> >> > > > >>>> state should specify whether the trx index is > empty > >>> or > >>> > > not. > >>> > > >> It > >>> > > >> >> > would > >>> > > >> >> > > > be > >>> > > >> >> > > > >>>> populated during segment archival. We would then > >>> iterate > >>> > > >> over > >>> > > >> >> the > >>> > > >> >> > > > >>>> metadata > >>> > > >> >> > > > >>>> for future segments without having to make a > remote > >>> call > >>> > > to > >>> > > >> >> > download > >>> > > >> >> > > > the > >>> > > >> >> > > > >>>> trx index itself. > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> The other solution for storing state at a > partition > >>> > level > >>> > > >> >> wouldn't > >>> > > >> >> > > > >>>> work, as > >>> > > >> >> > > > >>>> you mentioned, because we will have to change the > >>> state > >>> > on > >>> > > >> >> every > >>> > > >> >> > > > >>>> mutation > >>> > > >> >> > > > >>>> to the log i.e. at expiration of segments and > >>> append. > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> I have been thinking whether we can do something > >>> better > >>> > > than > >>> > > >> >> the > >>> > > >> >> > > > simple > >>> > > >> >> > > > >>>> solution, hence the delay in replying. Let me tell > >>> you > >>> > my > >>> > > >> half > >>> > > >> >> > baked > >>> > > >> >> > > > >>>> train > >>> > > >> >> > > > >>>> of thoughts, perhaps, you can explore this as > well. > >>> I > >>> > have > >>> > > >> been > >>> > > >> >> > > > thinking > >>> > > >> >> > > > >>>> about using LSO (last stable offset) to handle the > >>> case > >>> > > when > >>> > > >> >> the > >>> > > >> >> > > > >>>> partition > >>> > > >> >> > > > >>>> never had any transactions. For a partition which > >>> never > >>> > > had > >>> > > >> any > >>> > > >> >> > > > >>>> transaction, I would assume that the LSO is never > >>> > > >> initialized > >>> > > >> >> (or > >>> > > >> >> > is > >>> > > >> >> > > > >>>> equal > >>> > > >> >> > > > >>>> to log start offset)? Or is it equal to HW in that > >>> case? > >>> > > >> This > >>> > > >> >> is > >>> > > >> >> > > > >>>> something > >>> > > >> >> > > > >>>> that I am yet to verify. If this idea works, then > we > >>> > would > >>> > > >> not > >>> > > >> >> > have > >>> > > >> >> > > to > >>> > > >> >> > > > >>>> iterate through the metadata for the dominant case > >>> where > >>> > > the > >>> > > >> >> > > partition > >>> > > >> >> > > > >>>> had > >>> > > >> >> > > > >>>> no transactions at all. > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> -- > >>> > > >> >> > > > >>>> Divij Vaidya > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> On Tue, Jun 25, 2024 at 11:42 AM Kamal > >>> Chandraprakash < > >>> > > >> >> > > > >>>> kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>>> > Bump. Please review this proposal. > >>> > > >> >> > > > >>>> > > >>> > > >> >> > > > >>>> > > >>> > > >> >> > > > >>>> > On Mon, Jun 17, 2024 at 6:55 PM Kamal > >>> Chandraprakash < > >>> > > >> >> > > > >>>> > kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > > > >>>> > > >>> > > >> >> > > > >>>> > > Divij, > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > Thanks for the review! Updated the KIP with 1, > >>> 2, 3, > >>> > > >> and 4 > >>> > > >> >> > > review > >>> > > >> >> > > > >>>> > > comments. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > > 4. Potential alternative - Instead of having > >>> an > >>> > > >> algorithm > >>> > > >> >> > > where > >>> > > >> >> > > > we > >>> > > >> >> > > > >>>> > > traverse > >>> > > >> >> > > > >>>> > > across segment metadata and looking for > >>> > isTxnIdxEmpty > >>> > > >> flag, > >>> > > >> >> > > should > >>> > > >> >> > > > >>>> we > >>> > > >> >> > > > >>>> > > directly introduce a nextSegmentWithTrxInx() > >>> > function? > >>> > > >> This > >>> > > >> >> > > would > >>> > > >> >> > > > >>>> allow > >>> > > >> >> > > > >>>> > > implementers to optimize the otherwise linear > >>> scan > >>> > > >> across > >>> > > >> >> > > metadata > >>> > > >> >> > > > >>>> for > >>> > > >> >> > > > >>>> > all > >>> > > >> >> > > > >>>> > > segments by using techniques such as skip list > >>> etc. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > This is a good point to optimize the scan. We > >>> need > >>> > to > >>> > > >> >> maintain > >>> > > >> >> > > the > >>> > > >> >> > > > >>>> > > skip-list > >>> > > >> >> > > > >>>> > > for each leader-epoch. With unclean leader > >>> election, > >>> > > >> some > >>> > > >> >> > > brokers > >>> > > >> >> > > > >>>> may not > >>> > > >> >> > > > >>>> > > have > >>> > > >> >> > > > >>>> > > the complete lineage. This will expand the > >>> scope of > >>> > > the > >>> > > >> >> work. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > In this version, we plan to optimize only for > >>> the > >>> > > below > >>> > > >> 2 > >>> > > >> >> > cases: > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > 1. A partition does not have the transaction > >>> index > >>> > for > >>> > > >> any > >>> > > >> >> of > >>> > > >> >> > > the > >>> > > >> >> > > > >>>> > uploaded > >>> > > >> >> > > > >>>> > > segments. > >>> > > >> >> > > > >>>> > > The individual log segments `isTxnIdxEmpty` > >>> flag > >>> > > can > >>> > > >> be > >>> > > >> >> > > reduced > >>> > > >> >> > > > >>>> to a > >>> > > >> >> > > > >>>> > > single flag > >>> > > >> >> > > > >>>> > > in RLMM (using AND operator) that can serve > >>> the > >>> > > >> query - > >>> > > >> >> "Is > >>> > > >> >> > > all > >>> > > >> >> > > > >>>> the > >>> > > >> >> > > > >>>> > > transaction indexes empty for a partition?". > >>> > > >> >> > > > >>>> > > If yes, then we can directly scan the > >>> local-log > >>> > for > >>> > > >> >> aborted > >>> > > >> >> > > > >>>> > > transactions. > >>> > > >> >> > > > >>>> > > 2. A partition is produced using the > >>> transactional > >>> > > >> >> producer. > >>> > > >> >> > The > >>> > > >> >> > > > >>>> > > assumption made is that > >>> > > >> >> > > > >>>> > > the transaction will either > commit/rollback > >>> > within > >>> > > >> 15 > >>> > > >> >> > > minutes > >>> > > >> >> > > > >>>> > > (default transaction.max.timeout.ms = 15 > >>> mins), > >>> > > >> >> possibly > >>> > > >> >> > we > >>> > > >> >> > > > >>>> may have > >>> > > >> >> > > > >>>> > > to search only > >>> > > >> >> > > > >>>> > > a few consecutive remote log segments to > >>> collect > >>> > > the > >>> > > >> >> > aborted > >>> > > >> >> > > > >>>> > > transactions. > >>> > > >> >> > > > >>>> > > 3. A partition is being produced with both > >>> normal > >>> > and > >>> > > >> >> > > > transactional > >>> > > >> >> > > > >>>> > > producers. In this case, > >>> > > >> >> > > > >>>> > > we will be doing linear traversal. > >>> Maintaining a > >>> > > >> >> skip-list > >>> > > >> >> > > > might > >>> > > >> >> > > > >>>> > > improve the performance but > >>> > > >> >> > > > >>>> > > we delegate the RLMM implementation to > >>> users. If > >>> > > >> >> > implemented > >>> > > >> >> > > > >>>> > > incorrectly, then it can lead > >>> > > >> >> > > > >>>> > > to delivery of the aborted transaction > >>> records > >>> > to > >>> > > >> the > >>> > > >> >> > > > consumer. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > I notice two drawbacks with the reduction > >>> method as > >>> > > >> >> proposed > >>> > > >> >> > in > >>> > > >> >> > > > the > >>> > > >> >> > > > >>>> KIP: > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > 1. Even if one segment has a transaction > index, > >>> then > >>> > > we > >>> > > >> >> have > >>> > > >> >> > to > >>> > > >> >> > > > >>>> iterate > >>> > > >> >> > > > >>>> > > over all the metadata events. > >>> > > >> >> > > > >>>> > > 2. Assume that there are 10 segments and > >>> segment-5 > >>> > > has a > >>> > > >> >> txn > >>> > > >> >> > > > index. > >>> > > >> >> > > > >>>> Once > >>> > > >> >> > > > >>>> > > the first 6 segments are deleted, > >>> > > >> >> > > > >>>> > > due to breach by time/size/start-offset, > >>> then we > >>> > > >> should > >>> > > >> >> > > return > >>> > > >> >> > > > >>>> `true` > >>> > > >> >> > > > >>>> > > for "Is all the transaction indexes empty for > a > >>> > > >> partition?" > >>> > > >> >> > > > >>>> > > query but it will return `false` until the > >>> broker > >>> > > >> gets > >>> > > >> >> > > > restarted > >>> > > >> >> > > > >>>> and > >>> > > >> >> > > > >>>> > we > >>> > > >> >> > > > >>>> > > have to resort to iterate over all the > metadata > >>> > > events. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > > 5. Potential alternative#2 - We know that we > >>> may > >>> > > want > >>> > > >> the > >>> > > >> >> > > > indexes > >>> > > >> >> > > > >>>> of > >>> > > >> >> > > > >>>> > > multiple higher segments. Instead of fetching > >>> them > >>> > > >> >> > sequentially, > >>> > > >> >> > > > we > >>> > > >> >> > > > >>>> could > >>> > > >> >> > > > >>>> > > implement a parallel fetch or a pre-fetch for > >>> the > >>> > > >> indexes. > >>> > > >> >> > This > >>> > > >> >> > > > >>>> would > >>> > > >> >> > > > >>>> > help > >>> > > >> >> > > > >>>> > > hide the latency of sequentially fetching the > >>> trx > >>> > > >> indexes. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > We can implement parallel-fetch/prefetch once > >>> the > >>> > > tiered > >>> > > >> >> > storage > >>> > > >> >> > > > is > >>> > > >> >> > > > >>>> GAed. > >>> > > >> >> > > > >>>> > > Since this feature will be useful > >>> > > >> >> > > > >>>> > > to prefetch the next remote log segment and it > >>> > expands > >>> > > >> the > >>> > > >> >> > scope > >>> > > >> >> > > > of > >>> > > >> >> > > > >>>> the > >>> > > >> >> > > > >>>> > > work. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > > 6. Should the proposed API take "segmentId" > >>> as a > >>> > > >> >> parameter > >>> > > >> >> > > > >>>> instead of > >>> > > >> >> > > > >>>> > > "topicIdPartition"? Suggesting because > >>> isTxnIdEmpty > >>> > is > >>> > > >> not > >>> > > >> >> a > >>> > > >> >> > > > >>>> property of > >>> > > >> >> > > > >>>> > a > >>> > > >> >> > > > >>>> > > partition, instead it's a property of a > specific > >>> > > >> segment. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > We propose to use the `topicIdPartition` in > >>> > > >> >> > > > >>>> RemoteLogMetadataManager. > >>> > > >> >> > > > >>>> > > The implementation can fold/reduce the value > of > >>> the > >>> > > >> >> individual > >>> > > >> >> > > log > >>> > > >> >> > > > >>>> > segment > >>> > > >> >> > > > >>>> > > `isTxnIdEmpty` flag. This is added to avoid > >>> scanning > >>> > > all > >>> > > >> >> the > >>> > > >> >> > > > >>>> metadata > >>> > > >> >> > > > >>>> > > events > >>> > > >> >> > > > >>>> > > when the partition does not have a transaction > >>> index > >>> > > in > >>> > > >> >> any of > >>> > > >> >> > > the > >>> > > >> >> > > > >>>> > > segments. > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > On Mon, Jun 17, 2024 at 4:05 PM Divij Vaidya < > >>> > > >> >> > > > >>>> divijvaidy...@gmail.com> > >>> > > >> >> > > > >>>> > > wrote: > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > >> Hi Kamal > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> Thanks for bringing this up. This is a > problem > >>> > worth > >>> > > >> >> solving. > >>> > > >> >> > > We > >>> > > >> >> > > > >>>> have > >>> > > >> >> > > > >>>> > >> faced > >>> > > >> >> > > > >>>> > >> this in situations where some Kafka clients > >>> default > >>> > > to > >>> > > >> >> > > > >>>> read_committed > >>> > > >> >> > > > >>>> > mode > >>> > > >> >> > > > >>>> > >> and end up having high latencies for remote > >>> fetches > >>> > > >> due to > >>> > > >> >> > this > >>> > > >> >> > > > >>>> > traversal > >>> > > >> >> > > > >>>> > >> across all segments. > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> First some nits to clarify the KIP: > >>> > > >> >> > > > >>>> > >> 1. The motivation should make it clear that > >>> > traversal > >>> > > >> of > >>> > > >> >> all > >>> > > >> >> > > > >>>> segments is > >>> > > >> >> > > > >>>> > >> only in the worst case. If I am not mistaken > >>> > (please > >>> > > >> >> correct > >>> > > >> >> > me > >>> > > >> >> > > > if > >>> > > >> >> > > > >>>> > wrong), > >>> > > >> >> > > > >>>> > >> the traversal stops when it has found a > segment > >>> > > >> containing > >>> > > >> >> > LSO. > >>> > > >> >> > > > >>>> > >> 2. There is nothing like a non-txn topic. A > >>> > > transaction > >>> > > >> >> may > >>> > > >> >> > be > >>> > > >> >> > > > >>>> started > >>> > > >> >> > > > >>>> > on > >>> > > >> >> > > > >>>> > >> any topic. Perhaps, rephrase the statement in > >>> the > >>> > KIP > >>> > > >> so > >>> > > >> >> that > >>> > > >> >> > > it > >>> > > >> >> > > > is > >>> > > >> >> > > > >>>> > clear > >>> > > >> >> > > > >>>> > >> to the reader. > >>> > > >> >> > > > >>>> > >> 3. The hyperlink in the "the broker has to > >>> traverse > >>> > > all > >>> > > >> >> > the..." > >>> > > >> >> > > > >>>> seems > >>> > > >> >> > > > >>>> > >> incorrect. Did you want to point to > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > > >>> > > >> >> > > > >>> > > >> >> > > >>> > > >> >> > >>> > > >> > >>> > > > >>> > > >>> > https://github.com/apache/kafka/blob/21d60eabab8a14c8002611c65e092338bf584314/core/src/main/scala/kafka/log/LocalLog.scala#L444 > >>> > > >> >> > > > >>>> > >> ? > >>> > > >> >> > > > >>>> > >> 4. In the testing section, could we add a > test > >>> > plan? > >>> > > >> For > >>> > > >> >> > > > example, I > >>> > > >> >> > > > >>>> > would > >>> > > >> >> > > > >>>> > >> list down adding a test which would verify > the > >>> > number > >>> > > >> of > >>> > > >> >> > calls > >>> > > >> >> > > > >>>> made to > >>> > > >> >> > > > >>>> > >> RLMM. This test would have a higher number of > >>> calls > >>> > > >> >> earlier > >>> > > >> >> > vs. > >>> > > >> >> > > > >>>> after > >>> > > >> >> > > > >>>> > this > >>> > > >> >> > > > >>>> > >> KIP. > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> Other thoughts: > >>> > > >> >> > > > >>>> > >> 4. Potential alternative - Instead of having > an > >>> > > >> algorithm > >>> > > >> >> > where > >>> > > >> >> > > > we > >>> > > >> >> > > > >>>> > >> traverse > >>> > > >> >> > > > >>>> > >> across segment metadata and looking for > >>> > isTxnIdxEmpty > >>> > > >> >> flag, > >>> > > >> >> > > > should > >>> > > >> >> > > > >>>> we > >>> > > >> >> > > > >>>> > >> directly introduce a nextSegmentWithTrxInx() > >>> > > function? > >>> > > >> >> This > >>> > > >> >> > > would > >>> > > >> >> > > > >>>> allow > >>> > > >> >> > > > >>>> > >> implementers to optimize the otherwise linear > >>> scan > >>> > > >> across > >>> > > >> >> > > > metadata > >>> > > >> >> > > > >>>> for > >>> > > >> >> > > > >>>> > all > >>> > > >> >> > > > >>>> > >> segments by using techniques such as skip > list > >>> etc. > >>> > > >> >> > > > >>>> > >> 5. Potential alternative#2 - We know that we > >>> may > >>> > want > >>> > > >> the > >>> > > >> >> > > indexes > >>> > > >> >> > > > >>>> of > >>> > > >> >> > > > >>>> > >> multiple higher segments. Instead of fetching > >>> them > >>> > > >> >> > > sequentially, > >>> > > >> >> > > > we > >>> > > >> >> > > > >>>> > could > >>> > > >> >> > > > >>>> > >> implement a parallel fetch or a pre-fetch for > >>> the > >>> > > >> indexes. > >>> > > >> >> > This > >>> > > >> >> > > > >>>> would > >>> > > >> >> > > > >>>> > help > >>> > > >> >> > > > >>>> > >> hide the latency of sequentially fetching the > >>> trx > >>> > > >> indexes. > >>> > > >> >> > > > >>>> > >> 6. Should the proposed API take "segmentId" > as > >>> a > >>> > > >> parameter > >>> > > >> >> > > > instead > >>> > > >> >> > > > >>>> of > >>> > > >> >> > > > >>>> > >> "topicIdPartition"? Suggesting because > >>> isTxnIdEmpty > >>> > > is > >>> > > >> >> not a > >>> > > >> >> > > > >>>> property > >>> > > >> >> > > > >>>> > of a > >>> > > >> >> > > > >>>> > >> partition, instead it's a property of a > >>> specific > >>> > > >> segment. > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> Looking forward to hearing your thoughts > about > >>> the > >>> > > >> >> > > alternatives. > >>> > > >> >> > > > >>>> Let's > >>> > > >> >> > > > >>>> > get > >>> > > >> >> > > > >>>> > >> this fixed. > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> -- > >>> > > >> >> > > > >>>> > >> Divij Vaidya > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> On Mon, Jun 17, 2024 at 11:40 AM Kamal > >>> > > Chandraprakash < > >>> > > >> >> > > > >>>> > >> kamal.chandraprak...@gmail.com> wrote: > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > >> > Hi all, > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > I have opened a KIP-1058 > >>> > > >> >> > > > >>>> > >> > < > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > > >>> > > >> >> > > > >>> > > >> >> > > >>> > > >> >> > >>> > > >> > >>> > > > >>> > > >>> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-1058%3A+Txn+consumer+exerts+pressure+on+remote+storage+when+reading+non-txn+topic > >>> > > >> >> > > > >>>> > >> > > > >>> > > >> >> > > > >>>> > >> > to reduce the pressure on remote storage > when > >>> > > >> >> transactional > >>> > > >> >> > > > >>>> consumers > >>> > > >> >> > > > >>>> > >> are > >>> > > >> >> > > > >>>> > >> > reading non-txn topics from remote storage. > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > > >>> > > >> >> > > > >>> > > >> >> > > >>> > > >> >> > >>> > > >> > >>> > > > >>> > > >>> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-1058%3A+Txn+consumer+exerts+pressure+on+remote+storage+when+reading+non-txn+topic > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > Feedbacks and suggestions are welcome. > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > Thanks, > >>> > > >> >> > > > >>>> > >> > Kamal > >>> > > >> >> > > > >>>> > >> > > >>> > > >> >> > > > >>>> > >> > >>> > > >> >> > > > >>>> > > > >>> > > >> >> > > > >>>> > > >>> > > >> >> > > > >>>> > >>> > > >> >> > > > >>> > >>> > > >> >> > > > > >>> > > >> >> > > > >>> > > >> >> > > >>> > > >> >> > >>> > > >> > > >>> > > >> > >>> > > > > >>> > > > >>> > > >>> > >> >
