[jira] [Created] (KAFKA-13694) Some InvalidRecordException messages are thrown away

["RivenSun (Jira)" <ji...@apache.org>]

RivenSun created KAFKA-13694:
--------------------------------

             Summary: 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


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)