KAFKA-9206: throw a KafkaException when encountering CORRUPT_MESSAGE by agam · Pull Request #8111 · apache/kafka

agam · 2020-02-13T19:58:39Z

If the completed fetch has an error code signifying a corrupt message, throw a KafkaException that notes the offset and the topic-partition.
Also added a test that triggers the warning and verifies it is thrown.
Jira: https://issues.apache.org/jira/browse/KAFKA-9206

Committer Checklist (excluded from commit message)

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

agam · 2020-02-13T20:00:29Z

Note: I found LogCaptureAppender in the streams module to be useful for verifying log messages, and copied it to the clients module. Let me know if there's some common place I can place it instead.

ijuma · 2020-02-13T21:42:22Z

Why is this needed? Supposedly the slf4jlog4j dependency brings this too?

I guess you're doing it because we refer to log4j at compile time now?

yes, it's an import for LogCaptureAppender

hachikuji · 2020-02-14T02:49:15Z

I think the main question here is whether we want to raise this error to the user. If we happen to see message corruption in the fetched data, we raise that to the user, so should this case be different?

Ack. As discussed, will (1) change this to correspond to the current behavior for corruption in fetched data, i.e. throwing a KafkaException that wraps around a CorruptRecordException, and (2) remove the log message expectation.

Correction: just a KafkaException (the wrapping around I referred to was the text of the CorruptRecordException, not the exception itself)

…` after a fetch

agam · 2020-02-14T21:56:33Z

Updated, please take a look.

soondenana

i.e. throwing a KafkaException that wraps around a CorruptRecordException

Are you planing to have a CorruptRecordException exception too?

soondenana · 2020-02-14T22:17:55Z

+        // Trigger the exception.
+        assertThrows(KafkaException.class, () -> {
+            Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords =
+                    fetchedRecords();


nit: just calling fetchedRecords() should work here, or is compiler not liking it and forcing the return value assignment?

Agree, made the change to discard the return value.

agam · 2020-02-14T23:16:27Z

i.e. throwing a KafkaException that wraps around a CorruptRecordException

Are you planing to have a CorruptRecordException exception too?

No, clarified with a comment above too, I initially thought we were wrapping the exception in the "data corruption" case, but we're just wrapping its text.

hachikuji · 2020-02-16T23:28:22Z

            } else if (error == Errors.UNKNOWN_SERVER_ERROR) {
                log.warn("Unknown error fetching data for topic-partition {}", tp);
+            } else if (error == Errors.CORRUPT_MESSAGE) {
+                throw new KafkaException("Encountered corrupt message when fetching offset "


While we're at it, could we add the fetch offset to the IllegalStateException message below and the UNKNOWN_SERVER_ERROR log message above?

hachikuji · 2020-02-18T20:56:02Z

ok to test

agam · 2020-02-18T23:52:34Z

Test failure seems unrelated to this change: org.apache.kafka.clients.admin.KafkaAdminClientTest.testDefaultApiTimeoutOverride ?

(will rerun tests once)

agam · 2020-02-19T05:21:54Z

retest this please

hachikuji · 2020-02-19T17:20:19Z

The test failure is known flaky. I have a fix in #8101.

hachikuji · 2020-02-19T17:22:11Z

                log.debug("Received unknown leader epoch error in fetch for partition {}", tp);
            } else if (error == Errors.UNKNOWN_SERVER_ERROR) {
-                log.warn("Unknown error fetching data for topic-partition {}", tp);
+                log.warn("Unknown error fetching data while fetching offset "


nit: for log messages, can we use the {} placeholders?

hachikuji · 2020-02-19T17:51:58Z

retest this please

hachikuji · 2020-02-19T20:32:17Z

ok to test

agam · 2020-02-20T05:48:42Z

Couple of tests that failed in this run but not the previous run, likely flaky:

kafka.admin.DescribeConsumerGroupTest.testDescribeGroupMembersWithShortInitializationTimeout
kafka.admin.DescribeConsumerGroupTest.testDescribeGroupWithShortInitializationTimeout

hachikuji

LGTM. Thanks for the PR!

hachikuji · 2020-02-20T17:53:38Z

@agam One other small thing before we merge. Can you add CORRUPT_MESSAGE to the list of error codes in the javadoc for FetchResponse?

…Response`

hachikuji · 2020-02-20T22:05:10Z

retest this please

hachikuji · 2020-02-20T22:05:18Z

ok to test

hachikuji · 2020-02-21T00:49:17Z

retest this please

agam · 2020-02-21T16:19:19Z

Two test failures:

kafka.admin.DescribeConsumerGroupTest.testDescribeGroupMembersWithShortInitializationTimeout (same as earlier)
kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup (newer failure in this run)

@test

* apache-github/trunk: (23 commits) KAFKA-9530; Fix flaky test `testDescribeGroupWithShortInitializationTimeout` (apache#8154) HOTFIX: fix NPE in Kafka Streams IQ (apache#8158) MINOR: set scala version automatically based on gradle.properties KAFKA-9577; SaslClientAuthenticator incorrectly negotiates SASL_HANDSHAKE version (apache#8142) KAFKA-9441: Add internal TransactionManager (apache#8105) MINOR: Document endpoints for connector topic tracking (KIP-558) MINOR: Standby task commit needed when offsets updated (apache#8146) KAFKA-9206; Throw KafkaException on CORRUPT_MESSAGE error in Fetch response (apache#8111) MINOR: Remove unwanted regexReplace on tests/kafkatest/__init__.py KAFKA-9586: Fix errored json filename in ops documentation KAFKA-9575: Mention ZooKeeper 3.5.7 upgrade KAFKA-9481: Graceful handling TaskMigrated and TaskCorrupted (apache#8058) HOTFIX: don't try to remove uninitialized changelogs from assignment & don't prematurely mark task closed (apache#8140) MINOR: Fix javadoc at org.apache.kafka.clients.producer.KafkaProducer.InterceptorCallback#onCompletion (apache#7337) MINOR: Improve EOS example exception handling (apache#8052) MINOR: Fix a number of warnings in clients test (apache#8073) MINOR: Update shell scripts to support z/OS system (apache#7913) MINOR: Wording fix in Streams DSL docs (apache#5692) MINOR: Add missing @test annotation to MetadataTest#testMetadataMerge (apache#8141) KAFKA-9533: ValueTransform forwards `null` values (apache#8108) ...

@test

…etrics-common * confluent/master: (76 commits) KAFKA-9530; Fix flaky test `testDescribeGroupWithShortInitializationTimeout` (apache#8154) HOTFIX: fix NPE in Kafka Streams IQ (apache#8158) MINOR: set scala version automatically based on gradle.properties KAFKA-9577; SaslClientAuthenticator incorrectly negotiates SASL_HANDSHAKE version (apache#8142) KAFKA-9441: Add internal TransactionManager (apache#8105) MINOR: Document endpoints for connector topic tracking (KIP-558) MINOR: Standby task commit needed when offsets updated (apache#8146) Changes to migrate to Artifactory (#263) KAFKA-9206; Throw KafkaException on CORRUPT_MESSAGE error in Fetch response (apache#8111) MINOR: Remove unwanted regexReplace on tests/kafkatest/__init__.py KAFKA-9586: Fix errored json filename in ops documentation KAFKA-9575: Mention ZooKeeper 3.5.7 upgrade KAFKA-9481: Graceful handling TaskMigrated and TaskCorrupted (apache#8058) HOTFIX: don't try to remove uninitialized changelogs from assignment & don't prematurely mark task closed (apache#8140) MINOR: Fix javadoc at org.apache.kafka.clients.producer.KafkaProducer.InterceptorCallback#onCompletion (apache#7337) MINOR: Improve EOS example exception handling (apache#8052) MINOR: Fix a number of warnings in clients test (apache#8073) MINOR: Update shell scripts to support z/OS system (apache#7913) MINOR: Wording fix in Streams DSL docs (apache#5692) MINOR: Add missing @test annotation to MetadataTest#testMetadataMerge (apache#8141) ...

agam requested a review from hachikuji February 13, 2020 19:58

ijuma reviewed Feb 13, 2020

View reviewed changes

hachikuji reviewed Feb 14, 2020

View reviewed changes

Agam Brahma added 3 commits February 14, 2020 13:35

KAFKA-9206: throw a KafkaException when encountering `CORRUPT_MESSAGE…

efa1736

…` after a fetch

nit: formatting

969295d

added test

a58e1be

agam force-pushed the AK-9206 branch from 5946a84 to a58e1be Compare February 14, 2020 21:54

agam changed the title ~~KAFKA-9206: log a warning when encountering CORRUPT_MESSAGE~~ KAFKA-9206: throw a KafkaException when encountering CORRUPT_MESSAGE Feb 14, 2020

soondenana reviewed Feb 14, 2020

View reviewed changes

Review: throw away return value from fetchedRecords() in test

152323b

hachikuji reviewed Feb 16, 2020

View reviewed changes

Review: expanded log warnings to include fetch-offset

d8afd7e

hachikuji reviewed Feb 19, 2020

View reviewed changes

Review: use "{}" in log warnings

8cd5c55

hachikuji approved these changes Feb 20, 2020

View reviewed changes

Review: add reference to the CORRUPT_MESSAGE error in docs for `Fetch…

238eced

…Response`

hachikuji merged commit 84c4025 into apache:trunk Feb 21, 2020

agam deleted the AK-9206 branch February 21, 2020 19:35

Conversation

agam commented Feb 13, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

agam commented Feb 13, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agam commented Feb 14, 2020

Uh oh!

soondenana left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

agam commented Feb 14, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Feb 18, 2020

Uh oh!

agam commented Feb 18, 2020

Uh oh!

agam commented Feb 19, 2020

Uh oh!

hachikuji commented Feb 19, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Feb 19, 2020

Uh oh!

hachikuji commented Feb 19, 2020

Uh oh!

agam commented Feb 20, 2020

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Feb 20, 2020

Uh oh!

hachikuji commented Feb 20, 2020

Uh oh!

hachikuji commented Feb 20, 2020

Uh oh!

hachikuji commented Feb 21, 2020

Uh oh!

agam commented Feb 21, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

agam commented Feb 13, 2020 •

edited

Loading