Hi Chris, I am not sure if you are able to see the images I shared with you . Copying the code snippet below,
if (expectedRecordCount >= 0) { int missingCount = expectedRecordCount - (int) this.recordOffset() - 1; if (missingCount > 0) { if (transactionContext != null) { isMissedRecords = true; } else { throw new DataException(String.format("Missing %d records (expecting %d, actual %d)", missingCount, expectedRecordCount, this. recordOffset())); } } else if (missingCount < 0) { if (transactionContext != null) { isMissedRecords = true; } else { throw new DataException(String.format("Too many records (expecting %d, actual %d)", expectedRecordCount, this.recordOffset())); } } } addLastRecord(records, null, value); } //asn1 or binary abort if((config.parseErrorThreshold != null && parseErrorCount >= config. parseErrorThreshold && lastbatch && transactionContext != null) || (isMissedRecords && transactionContext != null && lastbatch)) { log.info("Transaction is aborting"); log.info("records = {}", records); if (!records.isEmpty()) { log.info("with record"); transactionContext.abortTransaction(records.get(records.size ()-1)); } else { log.info("without record"); transactionContext.abortTransaction(); } Thanks, Nitty On Wed, Mar 8, 2023 at 11:38 PM NITTY BENNY <nittybe...@gmail.com> wrote: > Hi Chris, > Sorry for the typo in my previous email. > > Regarding the point 2,* the task returns a batch of records from > SourceTask::poll (and, if using* > > > *the per-record API provided by the TransactionContext class, includes > atleast one record that should trigger a transaction commit/abort in > thatbatch)* > What if I am using the API without passing a record? We have 2 types of > use cases, one where on encountering an error record, we want to commit > previous successful batches and disregard the failed record and upcoming > batches. In this case we created the transactionContext just before reading > the file (file is our transaction boundary).Say if I have a file with 5 > records and batch size is 2, and in my 3rd batch I have one error record > then in that batch, I dont have a valid record to call commit or abort. But > I want to commit all the previous batches that were successfully parsed. > How do I do that? > > Second use case is where I want to abort a transaction if the record count > doesn't match. > Code Snippet : > [image: image.png] > There are no error records in this case. If you see I added the condition > of transactionContext check to implement exactly once, without > transaction it was just throwing the exception without calling the > addLastRecord() method and in the catch block it just logs the message and > return the list of records without the last record to poll().To make it > work, I called the method addLastRecord() in this case, so it is not > throwing the exception and list has last record as well. Then I called the > abort, everything got aborted. Why is abort not working without adding the > last record to the list? > [image: image.png] > > Code to call abort. > > > > > Thanks, > Nitty > > On Wed, Mar 8, 2023 at 4:26 PM Chris Egerton <chr...@aiven.io.invalid> > wrote: > >> Hi Nitty, >> >> I'm a little confused about what you mean by this part: >> >> > transaction is not getting completed because it is not commiting the >> transaction offest. >> >> The only conditions required for a transaction to be completed when a >> connector is defining its own transaction boundaries are: >> >> 1. The task requests a transaction commit/abort from the >> TransactionContext >> 2. The task returns a batch of records from SourceTask::poll (and, if >> using >> the per-record API provided by the TransactionContext class, includes at >> least one record that should trigger a transaction commit/abort in that >> batch) >> >> The Connect runtime should automatically commit source offsets to Kafka >> whenever a transaction is completed, either by commit or abort. This is >> because transactions should only be aborted for data that should never be >> re-read by the connector; if there is a validation error that should be >> handled by reconfiguring the connector, then the task should throw an >> exception instead of aborting the transaction. >> >> If possible, do you think you could provide a brief code snippet >> illustrating what your task is doing that's causing issues? >> >> Cheers, >> >> Chris (not Chrise 🙂) >> >> On Tue, Mar 7, 2023 at 10:17 AM NITTY BENNY <nittybe...@gmail.com> wrote: >> >> > Hi Chrise, >> > >> > Thanks for sharing the details. >> > >> > Regarding the use case, For Asn1 source connector we have a use case to >> > validate number of records in the file with the number of records in the >> > header. So currently, if validation fails we are not sending the last >> > record to the topic. But after introducing exactly once with connector >> > transaction boundary, I can see that if I call an abort when the >> validation >> > fails, transaction is not getting completed because it is not commiting >> the >> > transaction offest. I saw that transaction state changed to >> CompleteAbort. >> > So for my next transaction I am getting InvalidProducerEpochException >> and >> > then task stopped after that. I tried calling the abort after sending >> last >> > record to the topic then transaction getting completed. >> > >> > I dont know if I am doing anything wrong here. >> > >> > Please advise. >> > Thanks, >> > Nitty >> > >> > On Tue 7 Mar 2023 at 2:21 p.m., Chris Egerton <chr...@aiven.io.invalid> >> > wrote: >> > >> > > Hi Nitty, >> > > >> > > We've recently added some documentation on implementing exactly-once >> > source >> > > connectors here: >> > > >> > >> https://kafka.apache.org/documentation/#connect_exactlyoncesourceconnectors >> > > . >> > > To quote a relevant passage from those docs: >> > > >> > > > In order for a source connector to take advantage of this support, >> it >> > > must be able to provide meaningful source offsets for each record >> that it >> > > emits, and resume consumption from the external system at the exact >> > > position corresponding to any of those offsets without dropping or >> > > duplicating messages. >> > > >> > > So, as long as your source connector is able to use the Kafka Connect >> > > framework's offsets API correctly, it shouldn't be necessary to make >> any >> > > other code changes to the connector. >> > > >> > > To enable exactly-once support for source connectors on your Connect >> > > cluster, see the docs section here: >> > > https://kafka.apache.org/documentation/#connect_exactlyoncesource >> > > >> > > With regard to transactions, a transactional producer is always >> created >> > > automatically for your connector by the Connect runtime when >> exactly-once >> > > support is enabled on the worker. The only reason to set >> > > "transaction.boundary" to "connector" is if your connector would like >> to >> > > explicitly define its own transaction boundaries. In this case, it >> sounds >> > > like may be what you want; I just want to make sure to call out that >> in >> > > either case, you should not be directly instantiating a producer in >> your >> > > connector code, but let the Kafka Connect runtime do that for you, and >> > just >> > > worry about returning the right records from SourceTask::poll (and >> > possibly >> > > defining custom transactions using the TransactionContext API). >> > > >> > > With respect to your question about committing or aborting in certain >> > > circumstances, it'd be useful to know more about your use case, since >> it >> > > may not be necessary to define custom transaction boundaries in your >> > > connector at all. >> > > >> > > Cheers, >> > > >> > > Chris >> > > >> > > >> > > >> > > On Tue, Mar 7, 2023 at 7:21 AM NITTY BENNY <nittybe...@gmail.com> >> wrote: >> > > >> > > > Hi Team, >> > > > >> > > > Adding on top of this, I tried creating a TransactionContext object >> and >> > > > calling the commitTransaction and abortTranaction methods in source >> > > > connectors. >> > > > But the main problem I saw is that if there is any error while >> parsing >> > > the >> > > > record, connect is calling an abort but we have a use case to call >> > commit >> > > > in some cases. Is it a valid use case in terms of kafka connect? >> > > > >> > > > Another Question - Should I use a transactional producer instead >> > > > creating an object of TransactionContext? Below is the connector >> > > > configuration I am using. >> > > > >> > > > exactly.once.support: "required" >> > > > transaction.boundary: "connector" >> > > > >> > > > Could you please help me here? >> > > > >> > > > Thanks, >> > > > Nitty >> > > > >> > > > On Tue, Mar 7, 2023 at 12:29 AM NITTY BENNY <nittybe...@gmail.com> >> > > wrote: >> > > > >> > > > > Hi Team, >> > > > > I am trying to implement exactly once behavior in our source >> > connector. >> > > > Is >> > > > > there any sample source connector implementation available to >> have a >> > > look >> > > > > at? >> > > > > Regards, >> > > > > Nitty >> > > > > >> > > > >> > > >> > >> >