Re: Exactly once kafka connect query
Hi Chris, Is there any possibility to have a call with you? This is actually blocking our delivery, I actually want to sort with this. Thanks, Nitty On Mon, Mar 13, 2023 at 8:18 PM NITTY BENNY wrote: > Hi Chris, > > I really don't understand why a graceful shutdown will happen during a > commit operation? Am I understanding something wrong here?. I see > this happens when I have a batch of 2 valid records and in the second > batch the record is invalid. In that case I want to commit the valid > records. So I called commit and sent an empty list for the current batch to > poll() and then when the next file comes in and poll sees new records, I > see InvalidProducerEpochException. > Please advise me. > > Thanks, > Nitty > > On Mon, Mar 13, 2023 at 5:33 PM NITTY BENNY wrote: > >> Hi Chris, >> >> The difference is in the Task Classes, no difference for value/key >> convertors. >> >> I don’t see log messages for graceful shutdown. I am not clear on what >> you mean by shutting down the task. >> >> I called the commit operation for the successful records. Should I >> perform any other steps if I have an invalid record? >> Please advise. >> >> Thanks, >> Nitty >> >> On Mon, Mar 13, 2023 at 3:42 PM Chris Egerton >> wrote: >> >>> Hi Nitty, >>> >>> Thanks again for all the details here, especially the log messages. >>> >>> > The below mentioned issue is happening for Json connector only. Is >>> there >>> any difference with asn1,binary,csv and json connector? >>> >>> Can you clarify if the difference here is in the Connector/Task classens, >>> or if it's in the key/value converters that are configured for the >>> connector? The key/value converters are configured using the >>> "key.converter" and "value.converter" property and, if problems arise >>> with >>> them, the task will fail and, if it has a non-empty ongoing transaction, >>> that transaction will be automatically aborted since we close the task's >>> Kafka producer when it fails (or shuts down gracefully). >>> >>> With regards to these log messages: >>> >>> > org.apache.kafka.common.errors.ProducerFencedException: There is a >>> newer >>> producer with the same transactionalId which fences the current one. >>> >>> It looks like your tasks aren't shutting down gracefully in time, which >>> causes them to be fenced out by the Connect framework later on. Do you >>> see >>> messages like "Graceful stop of task failed" in the logs >>> for >>> your Connect worker? >>> >>> Cheers, >>> >>> Chris >>> >>> On Mon, Mar 13, 2023 at 10:58 AM NITTY BENNY >>> wrote: >>> >>> > Hi Chris, >>> > >>> > As you said, the below message is coming when I call an abort if there >>> is >>> > an invalid record, then for the next transaction I can see the below >>> > message and then the connector will be stopped. >>> > 2023-03-13 14:28:26,043 INFO [json-sftp-source-connector|task-0] >>> Aborting >>> > transaction for batch as requested by connector >>> > (org.apache.kafka.connect.runtime.ExactlyOnceWorkerSourceTask) >>> > [task-thread-json-sftp-source-connector-0] >>> > 2023-03-13 14:28:26,043 INFO [json-sftp-source-connector|task-0] >>> [Producer >>> > clientId=connector-producer-json-sftp-source-connector-0, >>> > transactionalId=connect-cluster-json-sftp-source-connector-0] Aborting >>> > incomplete transaction >>> (org.apache.kafka.clients.producer.KafkaProducer) >>> > [task-thread-json-sftp-source-connector-0] >>> > >>> > The issue with InvalidProducerEpoch is happening when I call the >>> commit if >>> > there is an invalid record, and in the next transaction I am getting >>> > InvalidProducerEpoch Exception and the messages are copied in the >>> previous >>> > email. I don't know if this will also be fixed by your bug fix.I am >>> using >>> > kafka 3.3.1 version as of now. >>> > >>> > Thanks, >>> > Nitty >>> > >>> > >>> > On Mon, Mar 13, 2023 at 10:47 AM NITTY BENNY >>> wrote: >>> > >>> > > Hi Chris, >>> > > >>> > > The below mentioned issue is happening for Json connector only. Is >>> there >>> > > any difference with asn1,binary,csv and json connector? >>> > > >>> > > Thanks, >>> > > Nitty >>> > > >>> > > On Mon, Mar 13, 2023 at 9:16 AM NITTY BENNY >>> > wrote: >>> > > >>> > >> Hi Chris, >>> > >> >>> > >> Sorry Chris, I am not able to reproduce the above issue. >>> > >> >>> > >> I want to share with you one more use case I found. >>> > >> The use case is in the first batch it returns 2 valid records and >>> then >>> > in >>> > >> the next batch it is an invalid record.Below is the >>> transaction_state >>> > >> topic, when I call a commit while processing an invalid record. >>> > >> >>> > >> >>> > >>> connect-cluster-json-sftp-source-connector-0::TransactionMetadata(transactionalId=connect-cluster-json-sftp-source-connector-0, >>> > >> producerId=11, producerEpoch=2, txnTimeoutMs=6, state=*Ongoing*, >>> > >> pendingState=None, topicPartitions=HashSet(streams-input-2), >>> > >> txnStartTimestamp=1678620463834, >>> txnLastUpdateTimestamp=1678620463834) >>> > >> >>
Re: Exactly once kafka connect query
Hi Nitty, Sorry, I should have clarified. The reason I'm thinking about shutdown here is that, when exactly-once support is enabled on a Kafka Connect cluster and a new set of task configurations is generated for a connector, the Connect framework makes an effort to shut down all the old task instances for that connector, and then fences out the transactional producers for all of those instances. I was thinking that this may lead to the producer exceptions you are seeing but, after double-checking this assumption, that does not appear to be the case. Would it be possible to share the source code for your connector and a reproduction scenario for what you're seeing? That may be easier than coordinating a call. Cheers, Chris On Tue, Mar 14, 2023 at 6:15 AM NITTY BENNY wrote: > Hi Chris, > > Is there any possibility to have a call with you? This is actually blocking > our delivery, I actually want to sort with this. > > Thanks, > Nitty > > On Mon, Mar 13, 2023 at 8:18 PM NITTY BENNY wrote: > > > Hi Chris, > > > > I really don't understand why a graceful shutdown will happen during a > > commit operation? Am I understanding something wrong here?. I see > > this happens when I have a batch of 2 valid records and in the second > > batch the record is invalid. In that case I want to commit the valid > > records. So I called commit and sent an empty list for the current batch > to > > poll() and then when the next file comes in and poll sees new records, I > > see InvalidProducerEpochException. > > Please advise me. > > > > Thanks, > > Nitty > > > > On Mon, Mar 13, 2023 at 5:33 PM NITTY BENNY > wrote: > > > >> Hi Chris, > >> > >> The difference is in the Task Classes, no difference for value/key > >> convertors. > >> > >> I don’t see log messages for graceful shutdown. I am not clear on what > >> you mean by shutting down the task. > >> > >> I called the commit operation for the successful records. Should I > >> perform any other steps if I have an invalid record? > >> Please advise. > >> > >> Thanks, > >> Nitty > >> > >> On Mon, Mar 13, 2023 at 3:42 PM Chris Egerton > >> wrote: > >> > >>> Hi Nitty, > >>> > >>> Thanks again for all the details here, especially the log messages. > >>> > >>> > The below mentioned issue is happening for Json connector only. Is > >>> there > >>> any difference with asn1,binary,csv and json connector? > >>> > >>> Can you clarify if the difference here is in the Connector/Task > classens, > >>> or if it's in the key/value converters that are configured for the > >>> connector? The key/value converters are configured using the > >>> "key.converter" and "value.converter" property and, if problems arise > >>> with > >>> them, the task will fail and, if it has a non-empty ongoing > transaction, > >>> that transaction will be automatically aborted since we close the > task's > >>> Kafka producer when it fails (or shuts down gracefully). > >>> > >>> With regards to these log messages: > >>> > >>> > org.apache.kafka.common.errors.ProducerFencedException: There is a > >>> newer > >>> producer with the same transactionalId which fences the current one. > >>> > >>> It looks like your tasks aren't shutting down gracefully in time, which > >>> causes them to be fenced out by the Connect framework later on. Do you > >>> see > >>> messages like "Graceful stop of task failed" in the logs > >>> for > >>> your Connect worker? > >>> > >>> Cheers, > >>> > >>> Chris > >>> > >>> On Mon, Mar 13, 2023 at 10:58 AM NITTY BENNY > >>> wrote: > >>> > >>> > Hi Chris, > >>> > > >>> > As you said, the below message is coming when I call an abort if > there > >>> is > >>> > an invalid record, then for the next transaction I can see the below > >>> > message and then the connector will be stopped. > >>> > 2023-03-13 14:28:26,043 INFO [json-sftp-source-connector|task-0] > >>> Aborting > >>> > transaction for batch as requested by connector > >>> > (org.apache.kafka.connect.runtime.ExactlyOnceWorkerSourceTask) > >>> > [task-thread-json-sftp-source-connector-0] > >>> > 2023-03-13 14:28:26,043 INFO [json-sftp-source-connector|task-0] > >>> [Producer > >>> > clientId=connector-producer-json-sftp-source-connector-0, > >>> > transactionalId=connect-cluster-json-sftp-source-connector-0] > Aborting > >>> > incomplete transaction > >>> (org.apache.kafka.clients.producer.KafkaProducer) > >>> > [task-thread-json-sftp-source-connector-0] > >>> > > >>> > The issue with InvalidProducerEpoch is happening when I call the > >>> commit if > >>> > there is an invalid record, and in the next transaction I am getting > >>> > InvalidProducerEpoch Exception and the messages are copied in the > >>> previous > >>> > email. I don't know if this will also be fixed by your bug fix.I am > >>> using > >>> > kafka 3.3.1 version as of now. > >>> > > >>> > Thanks, > >>> > Nitty > >>> > > >>> > > >>> > On Mon, Mar 13, 2023 at 10:47 AM NITTY BENNY > >>> wrote: > >>> > > >>> > > Hi Chris, > >>> > > > >>> > > The below mentioned issue is happe
Mirrormaker reverse offset syncing
Howdy, I'm currently using mirrormaker to translate message and offsets from cluster A -> B. If a consumer is currently processing messages from a mirrored topic, A.test_topic, in cluster B, is there anything mirroring the offsets back to cluster A for the original test_topic? i.e The consumer is up to offset 5 in A.test_topic in cluster B, is something translating back so that the consumer group is at offset 5 for test_topic in cluster A? If this doesn't happen within mirrormaker, are there any workarounds to get this behavior or would someone have to open up a KIP to add this functionality. Cheers, Chris
Invitation to participate in Fintech track at Community Over Code 2023
Dear Kafka Community, I hope you are doing well. As members of the ASF community, we share a common interest in open source and the innovative ways in which it can be leveraged to drive progress across various industries. In this spirit, I would like to invite you to participate in the Fintech track at Community Over Code 2023, the flagship event of the ASF (formerly known as ApacheCon). The Fintech track at Community Over Code 2023 allows ASF community experts to gather and share their knowledge and insights on the latest trends, challenges, and opportunities at the intersection of Fintech and open source. We are particularly interested in exploring the potential benefits of Apache Kafka for the Fintech industry. Its high throughput and real-time processing capabilities have the potential to revolutionize the way Fintech applications are built and deployed. We encourage you to share your expertise and insights on this topic or any other relevant to the intersection of Fintech and Apache Kafka. To submit your talk, visit https://communityovercode.org/call-for-presentations/ and follow the instructions on the page. The Call For Presentations (CFP) is open and will close on July 13th, 2023, at 23:59 UTC. We look forward to hearing from you and having you participate in the Fintech track at Community Over Code 2023. Best regards, Javier
The question about start kafka-server
Dear developer: Hello,get to the point,when I start kafka-server(kafka_2.13-3.1.0), I have encountered a prolem, I have put the relational files on attachment position, please help me to check the problems and give some solutions, I will be very happy for your help. Looking forward to your reply, thank you very much! Yours kafka_user;
