[ https://issues.apache.org/jira/browse/KAFKA-13694?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17499777#comment-17499777 ]
Guozhang Wang commented on KAFKA-13694: --------------------------------------- [~RivenSun] I think that's a good point, I agree with you generally about the solution. But one thing I think we need to be careful here is that, say if a single batch's all records fails the validation --- like, in your example, that producer forgot to set key for all records for a compacted topic --- then the `recordErrors: Seq[ApiRecordError]` could be very long. In practice, a batch could contain about a thousand or more records, so printing them all would flood the log entries. I think adding some heuristic tricks in the included errors would be preferred, e.g. (I just picked a magic number 3 below): ``` error.size() + " records have been rejected due to (only showing the first three errors): " + errors.subList(0, 3) ``` > Some InvalidRecordException messages are thrown away > ---------------------------------------------------- > > Key: KAFKA-13694 > URL: https://issues.apache.org/jira/browse/KAFKA-13694 > Project: Kafka > Issue Type: Improvement > Components: clients, core > Affects Versions: 3.0.0 > Reporter: RivenSun > Priority: Major > > 1.Example > Topic level config:"cleanup.policy":"compact" > But when the producer sends the message, the ProducerRecord does not specify > the key. > > producer.log > {code:java} > [kafka-producer-network-thread | producer-1] ERROR > us.zoom.mq.server.adapter.kafka.ProducerTest - the producer has a error:One > or more records have been rejected {code} > > > server.log > {code:java} > [2022-02-25 02:14:54,411] ERROR [ReplicaManager broker=1] Error processing > append operation on partition rivenTest4-0 (kafka.server.ReplicaManager) > org.apache.kafka.common.InvalidRecordException: One or more records have been > rejected {code} > Through the logs of the producer and server, we do not know the reason for > the failure of sending, only that the message was rejected by the server. > You can compare the RecordTooLargeException testCase, we can clearly know the > reason for the failure from the producer, and the server will not print the > log (the reason will be explained later) > producer_message_too_large.log : > {code:java} > [kafka-producer-network-thread | producer-1] ERROR > us.zoom.mq.server.adapter.kafka.ProducerTest - the producer has a error:The > request included a message larger than the max message size the server will > accept. > [kafka-producer-network-thread | producer-1] ERROR > us.zoom.mq.server.adapter.kafka.ProducerTest - the producer has a error:The > request included a message larger than the max message size the server will > accept. {code} > 2.RootCause > ReplicaManager#appendToLocalLog(...) -> > Partition#appendRecordsToLeader(...) -> > UnifiedLog#appendAsLeader(...) -> UnifiedLog#append(...) -> > LogValidator#validateMessagesAndAssignOffsets(...) > 1) Analyze the validateMessagesAndAssignOffsets method, > In the LogValidator#validateRecord method, validateKey and validateTimestamp > are called, and the error information of all messages is obtained: > Seq[ApiRecordError]; > In the subsequent processRecordErrors(recordErrors) method, currently only > special processing is done for Errors.INVALID_TIMESTAMP, because the ERROR > returned by validateKey is still the ordinary Errors.INVALID_RECORD, so the > code will run to > {code:java} > else { > throw new RecordValidationException(new InvalidRecordException( > "One or more records have been rejected"), errors) > }{code} > In fact, the *errors* variable here contains the specific information of each > recordError, but we did not put the errors information into the message of > InvalidRecordException. > 2).The exception thrown by processRecordErrors will be caught by > ReplicaManager#appendToLocalLog(...), we continue to analyze the > `catchException code` of appendToLocalLog. > Here, we can know the RecordTooLargeException, why the server does not print > the log. > Under case rve: RecordValidationException, > The server prints the log: processFailedRecord method, > and sends a response to the client: LogAppendResult method > In these two methods, we can find that we only use rve.invalidException, > For rve.recordErrors, the server neither prints it nor returns it to the > client. > 3.Solution > Two solutions, I prefer the second > 1)Similar to Errors.INVALID_TIMESTAMP, the validateKey method returns > Errors.INVALID_RECORD_WITHOUT_KEY, > In the processRecordErrors method, also do special processing for > Errors.INVALID_RECORD_WITHOUT_KEY > 2)Modify the logic of the processRecordErrors method, no longer distinguish > the types of Errors, and {*}Even if new INVALID_RECORD types will be added in > the future{*}, we uniformly return: > {code:java} > throw new RecordValidationException(new InvalidRecordException( > "One or more records have been rejected due to " + errors.toString()), > errors) {code} > Also need to add toString() method for ProduceResponse.RecordError class > {code:java} > @Override > public String toString() { > return "RecordError(" > + "batchIndex=" + batchIndex > + ", message=" + ((message == null) ? "null" : "'" + message + > "'") > + ")"; > } {code} > In the past, the toString method of ProduceResponse.PartitionResponse has > called the toString method of ProduceResponse.RecordError, *but before we > were missing the RecordError#toString method.* > -- This message was sent by Atlassian Jira (v8.20.1#820001)