You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ij...@apache.org on 2016/07/28 20:38:06 UTC

kafka git commit: MINOR: Consumer should throw KafkaException on invalid checksum

Repository: kafka
Updated Branches:
  refs/heads/trunk 6e8c6f17f -> 131f96868


MINOR: Consumer should throw KafkaException on invalid checksum

InvalidRecordException is not part of the public API so go back
to the behaviour before ff557f02 for now.

Author: Jason Gustafson <ja...@confluent.io>

Reviewers: Ismael Juma <is...@juma.me.uk>

Closes #1676 from hachikuji/raise-kafka-exception-on-invalid-crc


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/131f9686
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/131f9686
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/131f9686

Branch: refs/heads/trunk
Commit: 131f96868aa24f38f69ee2ab769ea3f289e5c1fc
Parents: 6e8c6f1
Author: Jason Gustafson <ja...@confluent.io>
Authored: Thu Jul 28 21:36:16 2016 +0100
Committer: Ismael Juma <is...@juma.me.uk>
Committed: Thu Jul 28 21:36:30 2016 +0100

----------------------------------------------------------------------
 .../org/apache/kafka/clients/consumer/internals/Fetcher.java     | 3 +--
 .../org/apache/kafka/clients/consumer/internals/FetcherTest.java | 4 ++--
 2 files changed, 3 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/131f9686/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
index c811a03..cf2ebc3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
@@ -39,7 +39,6 @@ import org.apache.kafka.common.metrics.stats.Max;
 import org.apache.kafka.common.metrics.stats.Rate;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
-import org.apache.kafka.common.record.InvalidRecordException;
 import org.apache.kafka.common.record.LogEntry;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.Record;
@@ -610,7 +609,7 @@ public class Fetcher<K, V> {
         Record record = logEntry.record();
 
         if (this.checkCrcs && !record.isValid())
-            throw new InvalidRecordException("Record for partition " + partition + " at offset "
+            throw new KafkaException("Record for partition " + partition + " at offset "
                     + logEntry.offset() + " is corrupt (stored crc = " + record.checksum()
                     + ", computed crc = "
                     + record.computeChecksum()

http://git-wip-us.apache.org/repos/asf/kafka/blob/131f9686/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
index 2fbd43e..ba04cb5 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java
@@ -25,6 +25,7 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.clients.consumer.OffsetOutOfRangeException;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
@@ -40,7 +41,6 @@ import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.protocol.types.Struct;
 import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.record.Compressor;
-import org.apache.kafka.common.record.InvalidRecordException;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.Record;
 import org.apache.kafka.common.requests.FetchRequest;
@@ -219,7 +219,7 @@ public class FetcherTest {
         try {
             fetcher.fetchedRecords();
             fail("fetchedRecords should have raised");
-        } catch (InvalidRecordException e) {
+        } catch (KafkaException e) {
             // the position should not advance since no data has been returned
             assertEquals(0, subscriptions.position(tp).longValue());
         }