KAFKA-17104: fix InvalidMessageCrcRecordsPerSec is not updated in validating Legac…

[TaiJuWu]

…yRecord

*More detailed description of your change,

*Summary of testing strategy (including rationale)
Uint test

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[chia7712]

@TaiJuWu could you please update docs ?

kafka/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java

Line 127 in ec9dabf

* Throw an InvalidRecordException if isValid is false for this record

[TaiJuWu]

@TaiJuWu could you please update docs ?

kafka/clients/src/main/java/org/apache/kafka/common/record/LegacyRecord.java

Line 127 in ec9dabf

* Throw an InvalidRecordException if isValid is false for this record

Done. Thanks for your check.

[chia7712]

@TaiJuWu please rebase code to trigger QA again

[TaiJuWu]

@TaiJuWu please rebase code to trigger QA again

Hi @chia7712 , It seems not CI issue because it appear two times.
I guess number from CsvSource can't be converted into Byte directly with old scala/java version.
Or maybe there are other reasons.
I will build environment to test it if needed.

Or addess this in #16167 ?
WDYT?

[chia7712]

I guess number from CsvSource can't be converted into Byte directly with old scala/java version.
Or maybe there are other reasons.
I will build environment to test it if needed.

not really. The deadlock is caused by testLogRecoveryMetrics, and it will be fixed by #16614

[TaiJuWu]

I guess number from CsvSource can't be converted into Byte directly with old scala/java version.
Or maybe there are other reasons.
I will build environment to test it if needed.

not really. The deadlock is caused by testLogRecoveryMetrics, and it will be fixed by #16614

OK. Thanks your review and information.

[chia7712]

In order to keep consistency, we should rethrow CorruptRecordException here. for example:

            } catch (CorruptRecordException e) {
                metricsRecorder.recordInvalidChecksums();
                throw e;
            } catch (InvalidRecordException e) {
                metricsRecorder.recordInvalidChecksums();
                throw new CorruptRecordException(e.getMessage() + " in topic partition " + topicPartition);
            }

[TaiJuWu]

Done. Thanks for this suggestion.

[chia7712]

LGTM