[ https://issues.apache.org/jira/browse/KAFKA-13694?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17498517#comment-17498517 ]
RivenSun commented on KAFKA-13694: ---------------------------------- Hi [~guozhang] Thanks for your reply, I agree with you totally. I will create a PR later, I hope you can help to review it,thanks. There is a branch in the code of *MessageDataGenerator#generateFieldToString() method* that deals with fields of type *String :* {code:java} else if (field.type().isString()) { buffer.printf("+ \"%s%s=\" + ((%s == null) ? \"null\" : \"'\" + %s.toString() + \"'\")%n", prefix, field.camelCaseName(), field.camelCaseName(), field.camelCaseName()); } {code} For fields of type String, the toString() method is still called, which brings an unnecessary method call. You can see their toString method in all xxx.java in clients/src/generated/java directory: example: AddOffsetsToTxnRequestData#toString() {code:java} @Override public String toString() { return "AddOffsetsToTxnRequestData(" + "transactionalId=" + ((transactionalId == null) ? "null" : "'" + transactionalId.toString() + "'") + ", producerId=" + producerId + ", producerEpoch=" + producerEpoch + ", groupId=" + ((groupId == null) ? "null" : "'" + groupId.toString() + "'") + ")"; } {code} [~cmccabe] [~guozhang] What do you think? Thanks. > 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)