Repository: kafka Updated Branches: refs/heads/0.10.2 d8f9e18aa -> 8812d151a
KAFKA-4630; Implement RecordTooLargeException when communicating with pre-KIP-74 brokers Author: Colin P. Mccabe <[email protected]> Reviewers: Jason Gustafson <[email protected]>, Ismael Juma <[email protected]> Closes #2390 from cmccabe/KAFKA-4630 (cherry picked from commit 5b4e299a46a8ee92e450c54880408713606e4f52) Signed-off-by: Ismael Juma <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/8812d151 Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/8812d151 Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/8812d151 Branch: refs/heads/0.10.2 Commit: 8812d151a9cb7d3a71fdb7c170a661ebf5214358 Parents: d8f9e18 Author: Colin P. Mccabe <[email protected]> Authored: Thu Jan 26 09:24:17 2017 +0000 Committer: Ismael Juma <[email protected]> Committed: Thu Jan 26 11:23:54 2017 +0000 ---------------------------------------------------------------------- .../apache/kafka/clients/NodeApiVersions.java | 37 +++++++++ .../clients/consumer/internals/Fetcher.java | 82 +++++++++++--------- .../kafka/common/requests/FetchRequest.java | 1 + .../kafka/common/requests/FetchResponse.java | 6 ++ .../org/apache/kafka/clients/MockClient.java | 7 +- .../kafka/clients/NodeApiVersionsTest.java | 3 +- .../clients/consumer/internals/FetcherTest.java | 58 ++++++++++++++ .../admin/BrokerApiVersionsCommandTest.scala | 2 +- .../client_compatibility_features_test.py | 4 + .../kafka/tools/ClientCompatibilityTest.java | 62 +++++++++------ 10 files changed, 198 insertions(+), 64 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java b/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java index b90009b..1c5d8fb 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java +++ b/clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java @@ -19,7 +19,10 @@ import org.apache.kafka.common.requests.ApiVersionsResponse.ApiVersion; import org.apache.kafka.common.utils.Utils; import java.util.Collection; +import java.util.Collections; import java.util.EnumMap; +import java.util.LinkedList; +import java.util.List; import java.util.Map; import java.util.NoSuchElementException; import java.util.TreeMap; @@ -30,6 +33,40 @@ public class NodeApiVersions { // An array of the usable versions of each API, indexed by ApiKeys ID. private final Map<ApiKeys, Short> usableVersions = new EnumMap<>(ApiKeys.class); + /** + * Create a NodeApiVersions object with the current ApiVersions. + * + * @return A new NodeApiVersions object. + */ + public static NodeApiVersions create() { + return create(Collections.EMPTY_LIST); + } + + /** + * Create a NodeApiVersions object. + * + * @param overrides API versions to override. Any ApiVersion not specified here will be set to the current client + * value. + * @return A new NodeApiVersions object. + */ + public static NodeApiVersions create(Collection<ApiVersion> overrides) { + List<ApiVersion> apiVersions = new LinkedList<>(overrides); + for (ApiKeys apiKey : ApiKeys.values()) { + boolean exists = false; + for (ApiVersion apiVersion : apiVersions) { + if (apiVersion.apiKey == apiKey.id) { + exists = true; + break; + } + } + if (!exists) { + apiVersions.add(new ApiVersion(apiKey.id, ProtoUtils.oldestVersion(apiKey.id), + ProtoUtils.latestVersion(apiKey.id))); + } + } + return new NodeApiVersions(apiVersions); + } + public NodeApiVersions(Collection<ApiVersion> apiVersions) { this.apiVersions = apiVersions; for (ApiVersion apiVersion : apiVersions) { http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/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 6bd454e..af82570 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 @@ -15,6 +15,7 @@ package org.apache.kafka.clients.consumer.internals; import org.apache.kafka.clients.ClientResponse; import org.apache.kafka.clients.Metadata; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.NoOffsetForPartitionException; import org.apache.kafka.clients.consumer.OffsetAndTimestamp; @@ -27,6 +28,7 @@ import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.InvalidMetadataException; import org.apache.kafka.common.errors.InvalidTopicException; +import org.apache.kafka.common.errors.RecordTooLargeException; import org.apache.kafka.common.errors.RetriableException; import org.apache.kafka.common.errors.SerializationException; import org.apache.kafka.common.errors.TimeoutException; @@ -174,7 +176,8 @@ public class Fetcher<K, V> implements SubscriptionState.Listener { TopicPartition partition = entry.getKey(); long fetchOffset = request.fetchData().get(partition).offset; FetchResponse.PartitionData fetchData = entry.getValue(); - completedFetches.add(new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator)); + completedFetches.add(new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator, + request.version())); } sensors.fetchLatency.record(resp.requestLatencyMs()); @@ -334,44 +337,24 @@ public class Fetcher<K, V> implements SubscriptionState.Listener { private void resetOffset(TopicPartition partition) { OffsetResetStrategy strategy = subscriptions.resetStrategy(partition); log.debug("Resetting offset for partition {} to {} offset.", partition, strategy.name().toLowerCase(Locale.ROOT)); - long offset = listOffset(partition, strategy); + final long timestamp; + if (strategy == OffsetResetStrategy.EARLIEST) + timestamp = ListOffsetRequest.EARLIEST_TIMESTAMP; + else if (strategy == OffsetResetStrategy.LATEST) + timestamp = ListOffsetRequest.LATEST_TIMESTAMP; + else + throw new NoOffsetForPartitionException(partition); + Map<TopicPartition, OffsetAndTimestamp> offsetsByTimes = retrieveOffsetsByTimes( + Collections.singletonMap(partition, timestamp), Long.MAX_VALUE, false); + OffsetAndTimestamp offsetAndTimestamp = offsetsByTimes.get(partition); + if (offsetAndTimestamp == null) + throw new NoOffsetForPartitionException(partition); + long offset = offsetAndTimestamp.offset(); // we might lose the assignment while fetching the offset, so check it is still active if (subscriptions.isAssigned(partition)) this.subscriptions.seek(partition, offset); } - private long listOffset(TopicPartition partition, OffsetResetStrategy strategy) { - final long timestamp; - switch (strategy) { - case EARLIEST: - timestamp = ListOffsetRequest.EARLIEST_TIMESTAMP; - break; - case LATEST: - timestamp = ListOffsetRequest.LATEST_TIMESTAMP; - break; - default: - throw new NoOffsetForPartitionException(partition); - } - while (true) { - RequestFuture<Map<TopicPartition, OffsetAndTimestamp>> future = - sendListOffsetRequests(false, Collections.singletonMap(partition, timestamp)); - client.poll(future); - if (future.succeeded()) { - OffsetAndTimestamp offsetAndTimestamp = future.value().get(partition); - if (offsetAndTimestamp == null) - throw new NoOffsetForPartitionException(partition); - return offsetAndTimestamp.offset(); - } - if (!future.isRetriable()) - throw future.exception(); - if (future.exception() instanceof InvalidMetadataException) - client.awaitMetadataUpdate(); - else - time.sleep(retryBackoffMs); - } - } - - public Map<TopicPartition, OffsetAndTimestamp> getOffsetsByTimes(Map<TopicPartition, Long> timestampsToSearch, long timeout) { return retrieveOffsetsByTimes(timestampsToSearch, timeout, true); @@ -636,7 +619,7 @@ public class Fetcher<K, V> implements SubscriptionState.Listener { } else { offset = partitionData.offsets.get(0); } - log.debug("Handling v0 ListOffsetResponse response for {}. Fetched offset {}", + log.debug("Handling v0 ListOffsetResponse response for {}. Fetched offset {}", topicPartition, offset); if (offset != ListOffsetResponse.UNKNOWN_OFFSET) { OffsetAndTimestamp offsetAndTimestamp = new OffsetAndTimestamp(offset, -1); @@ -644,7 +627,7 @@ public class Fetcher<K, V> implements SubscriptionState.Listener { } } else { // Handle v1 and later response - log.debug("Handling ListOffsetResponse response for {}. Fetched offset {}, timestamp {}", + log.debug("Handling ListOffsetResponse response for {}. Fetched offset {}, timestamp {}", topicPartition, partitionData.offset, partitionData.timestamp); if (partitionData.offset != ListOffsetResponse.UNKNOWN_OFFSET) { OffsetAndTimestamp offsetAndTimestamp = @@ -754,12 +737,14 @@ public class Fetcher<K, V> implements SubscriptionState.Listener { } List<ConsumerRecord<K, V>> parsed = new ArrayList<>(); + boolean skippedRecords = false; for (LogEntry logEntry : partition.records.deepEntries()) { // Skip the messages earlier than current position. if (logEntry.offset() >= position) { parsed.add(parseRecord(tp, logEntry)); bytes += logEntry.sizeInBytes(); - } + } else + skippedRecords = true; } recordsCount = parsed.size(); @@ -767,6 +752,24 @@ public class Fetcher<K, V> implements SubscriptionState.Listener { log.trace("Adding fetched record for partition {} with offset {} to buffered record list", tp, position); parsedRecords = new PartitionRecords<>(fetchOffset, tp, parsed); + if (parsed.isEmpty() && !skippedRecords && (partition.records.sizeInBytes() > 0)) { + if (completedFetch.responseVersion < 3) { + // Implement the pre KIP-74 behavior of throwing a RecordTooLargeException. + Map<TopicPartition, Long> recordTooLargePartitions = Collections.singletonMap(tp, fetchOffset); + throw new RecordTooLargeException("There are some messages at [Partition=Offset]: " + + recordTooLargePartitions + " whose size is larger than the fetch size " + this.fetchSize + + " and hence cannot be returned. Please considering upgrading your broker to 0.10.1.0 or " + + "newer to avoid this issue. Alternately, increase the fetch size on the client (using " + + ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG + ")", + recordTooLargePartitions); + } else { + // This should not happen with brokers that support FetchRequest/Response V3 or higher (i.e. KIP-74) + throw new KafkaException("Failed to make progress reading messages at " + tp + "=" + + fetchOffset + ". Received a non-empty fetch response from the server, but no " + + "complete records were found."); + } + } + if (partition.highWatermark >= 0) { log.trace("Received {} records in fetch response for partition {} with offset {}", parsed.size(), tp, position); subscriptions.updateHighWatermark(tp, partition.highWatermark); @@ -894,15 +897,18 @@ public class Fetcher<K, V> implements SubscriptionState.Listener { private final long fetchedOffset; private final FetchResponse.PartitionData partitionData; private final FetchResponseMetricAggregator metricAggregator; + private final short responseVersion; private CompletedFetch(TopicPartition partition, long fetchedOffset, FetchResponse.PartitionData partitionData, - FetchResponseMetricAggregator metricAggregator) { + FetchResponseMetricAggregator metricAggregator, + short responseVersion) { this.partition = partition; this.fetchedOffset = fetchedOffset; this.partitionData = partitionData; this.metricAggregator = metricAggregator; + this.responseVersion = responseVersion; } } http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java index 4768a20..5700e9e 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java @@ -142,6 +142,7 @@ public class FetchRequest extends AbstractRequest { append(", replicaId=").append(replicaId). append(", maxWait=").append(maxWait). append(", minBytes=").append(minBytes). + append(", maxBytes=").append(maxBytes). append(", fetchData=").append(Utils.mkString(fetchData)). append(")"); return bld.toString(); http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java index 66b01a0..965b207 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java +++ b/clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java @@ -82,6 +82,12 @@ public class FetchResponse extends AbstractResponse { this.highWatermark = highWatermark; this.records = records; } + + @Override + public String toString() { + return "(errorCode=" + errorCode + ", highWaterMark=" + highWatermark + + ", records=" + records + ")"; + } } /** http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/clients/src/test/java/org/apache/kafka/clients/MockClient.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/MockClient.java b/clients/src/test/java/org/apache/kafka/clients/MockClient.java index bfde9f9..50ed131 100644 --- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java +++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java @@ -73,6 +73,7 @@ public class MockClient implements KafkaClient { private final Queue<ClientResponse> responses = new ConcurrentLinkedDeque<>(); private final Queue<FutureResponse> futureResponses = new ArrayDeque<>(); private final Queue<Cluster> metadataUpdates = new ArrayDeque<>(); + private volatile NodeApiVersions nodeApiVersions = NodeApiVersions.create(); public MockClient(Time time) { this.time = time; @@ -145,7 +146,8 @@ public class MockClient implements KafkaClient { FutureResponse futureResp = iterator.next(); if (futureResp.node != null && !request.destination().equals(futureResp.node.idString())) continue; - + request.requestBuilder().setVersion(nodeApiVersions.usableVersion( + request.requestBuilder().apiKey())); AbstractRequest abstractRequest = request.requestBuilder().build(); if (!futureResp.requestMatcher.matches(abstractRequest)) throw new IllegalStateException("Next in line response did not match expected request"); @@ -334,4 +336,7 @@ public class MockClient implements KafkaClient { boolean matches(AbstractRequest body); } + public void setNodeApiVersions(NodeApiVersions nodeApiVersions) { + this.nodeApiVersions = nodeApiVersions; + } } http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java b/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java index 861a28f..f72137b 100644 --- a/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/NodeApiVersionsTest.java @@ -33,8 +33,7 @@ public class NodeApiVersionsTest { @Test public void testUnsupportedVersionsToString() { - NodeApiVersions versions = new NodeApiVersions( - Collections.<ApiVersion>emptyList()); + NodeApiVersions versions = new NodeApiVersions(Collections.<ApiVersion>emptyList()); StringBuilder bld = new StringBuilder(); String prefix = "("; for (ApiKeys apiKey : ApiKeys.values()) { http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/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 210af6d..ecaade2 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 @@ -18,6 +18,7 @@ package org.apache.kafka.clients.consumer.internals; import org.apache.kafka.clients.Metadata; import org.apache.kafka.clients.MockClient; +import org.apache.kafka.clients.NodeApiVersions; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.OffsetAndMetadata; @@ -31,11 +32,13 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.InvalidTopicException; +import org.apache.kafka.common.errors.RecordTooLargeException; import org.apache.kafka.common.errors.SerializationException; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.TopicAuthorizationException; import org.apache.kafka.common.metrics.KafkaMetric; import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.record.ByteBufferOutputStream; import org.apache.kafka.common.record.CompressionType; @@ -44,6 +47,7 @@ import org.apache.kafka.common.record.MemoryRecordsBuilder; import org.apache.kafka.common.record.Record; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.requests.AbstractRequest; +import org.apache.kafka.common.requests.ApiVersionsResponse; import org.apache.kafka.common.requests.FetchRequest; import org.apache.kafka.common.requests.FetchResponse; import org.apache.kafka.common.requests.ListOffsetRequest; @@ -316,6 +320,60 @@ public class FetcherTest { assertEquals(30L, consumerRecords.get(2).offset()); } + /** + * Test the case where the client makes a pre-v3 FetchRequest, but the server replies with only a partial + * request. This happens when a single message is larger than the per-partition limit. + */ + @Test + public void testFetchRequestWhenRecordTooLarge() { + try { + client.setNodeApiVersions(NodeApiVersions.create(Collections.singletonList( + new ApiVersionsResponse.ApiVersion(ApiKeys.FETCH.id, (short) 2, (short) 2)))); + makeFetchRequestWithIncompleteRecord(); + try { + fetcher.fetchedRecords(); + fail("RecordTooLargeException should have been raised"); + } catch (RecordTooLargeException e) { + assertTrue(e.getMessage().startsWith("There are some messages at [Partition=Offset]: ")); + // the position should not advance since no data has been returned + assertEquals(0, subscriptions.position(tp).longValue()); + } + } finally { + client.setNodeApiVersions(NodeApiVersions.create()); + } + } + + /** + * Test the case where the client makes a post KIP-74 FetchRequest, but the server replies with only a + * partial request. For v3 and later FetchRequests, the implementation of KIP-74 changed the behavior + * so that at least one message is always returned. Therefore, this case should not happen, and it indicates + * that an internal error has taken place. + */ + @Test + public void testFetchRequestInternalError() { + makeFetchRequestWithIncompleteRecord(); + try { + fetcher.fetchedRecords(); + fail("RecordTooLargeException should have been raised"); + } catch (KafkaException e) { + assertTrue(e.getMessage().startsWith("Failed to make progress reading messages")); + // the position should not advance since no data has been returned + assertEquals(0, subscriptions.position(tp).longValue()); + } + } + + private void makeFetchRequestWithIncompleteRecord() { + subscriptions.assignFromUser(singleton(tp)); + subscriptions.seek(tp, 0); + assertEquals(1, fetcher.sendFetches()); + assertFalse(fetcher.hasCompletedFetches()); + MemoryRecords partialRecord = MemoryRecords.readableRecords( + ByteBuffer.wrap(new byte[]{0, 0, 0, 0, 0, 0, 0, 0})); + client.prepareResponse(fetchResponse(partialRecord, Errors.NONE.code(), 100L, 0)); + consumerClient.poll(0); + assertTrue(fetcher.hasCompletedFetches()); + } + @Test public void testUnauthorizedTopic() { subscriptions.assignFromUser(singleton(tp)); http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala ---------------------------------------------------------------------- diff --git a/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala b/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala index ff93f22..8fa489d 100644 --- a/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala +++ b/core/src/test/scala/integration/kafka/admin/BrokerApiVersionsCommandTest.scala @@ -42,7 +42,7 @@ class BrokerApiVersionsCommandTest extends KafkaServerTestHarness { val lineIter = content.split("\n").iterator assertTrue(lineIter.hasNext) assertEquals(s"$brokerList (id: 0 rack: null) -> (", lineIter.next) - val nodeApiVersions = new NodeApiVersions(ApiVersionsResponse.API_VERSIONS_RESPONSE.apiVersions) + val nodeApiVersions = NodeApiVersions.create for (apiKey <- ApiKeys.values) { val apiVersion = nodeApiVersions.apiVersion(apiKey) val versionRangeStr = http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/tests/kafkatest/tests/client/client_compatibility_features_test.py ---------------------------------------------------------------------- diff --git a/tests/kafkatest/tests/client/client_compatibility_features_test.py b/tests/kafkatest/tests/client/client_compatibility_features_test.py index 3c96b84..36d76b4 100644 --- a/tests/kafkatest/tests/client/client_compatibility_features_test.py +++ b/tests/kafkatest/tests/client/client_compatibility_features_test.py @@ -30,9 +30,11 @@ def get_broker_features(broker_version): if (broker_version < V_0_10_1_0): features["offsets-for-times-supported"] = False features["cluster-id-supported"] = False + features["expect-record-too-large-exception"] = True else: features["offsets-for-times-supported"] = True features["cluster-id-supported"] = True + features["expect-record-too-large-exception"] = False return features def run_command(node, cmd, ssh_log_file): @@ -74,10 +76,12 @@ class ClientCompatibilityFeaturesTest(Test): "--bootstrap-server %s " "--offsets-for-times-supported %s " "--cluster-id-supported %s " + "--expect-record-too-large-exception %s " "--topic %s " % (self.zk.path.script("kafka-run-class.sh", node), self.kafka.bootstrap_servers(), features["offsets-for-times-supported"], features["cluster-id-supported"], + features["expect-record-too-large-exception"], self.topics.keys()[0])) results_dir = TestContext.results_dir(self.test_context, 0) os.makedirs(results_dir) http://git-wip-us.apache.org/repos/asf/kafka/blob/8812d151/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java ---------------------------------------------------------------------- diff --git a/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java b/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java index 69202a0..d7eb717 100644 --- a/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java +++ b/tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java @@ -22,6 +22,7 @@ import net.sourceforge.argparse4j.ArgumentParsers; import net.sourceforge.argparse4j.inf.ArgumentParser; import net.sourceforge.argparse4j.inf.ArgumentParserException; import net.sourceforge.argparse4j.inf.Namespace; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; @@ -35,6 +36,7 @@ import org.apache.kafka.common.ClusterResourceListener; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.ObsoleteBrokerException; +import org.apache.kafka.common.errors.RecordTooLargeException; import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.utils.Time; @@ -66,12 +68,14 @@ public class ClientCompatibilityTest { final String topic; final boolean offsetsForTimesSupported; final boolean expectClusterId; + final boolean expectRecordTooLargeException; TestConfig(Namespace res) { this.bootstrapServer = res.getString("bootstrapServer"); this.topic = res.getString("topic"); this.offsetsForTimesSupported = res.getBoolean("offsetsForTimesSupported"); this.expectClusterId = res.getBoolean("clusterIdSupported"); + this.expectRecordTooLargeException = res.getBoolean("expectRecordTooLargeException"); } } @@ -108,6 +112,15 @@ public class ClientCompatibilityTest { .dest("clusterIdSupported") .metavar("CLUSTER_ID_SUPPORTED") .help("True if cluster IDs are supported. False if cluster ID always appears as null."); + parser.addArgument("--expect-record-too-large-exception") + .action(store()) + .required(true) + .type(Boolean.class) + .dest("expectRecordTooLargeException") + .metavar("EXPECT_RECORD_TOO_LARGE_EXCEPTION") + .help("True if we should expect a RecordTooLargeException when trying to read from a topic " + + "that contains a message that is bigger than " + ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG + + ". This is pre-KIP-74 behavior."); Namespace res = null; try { res = parser.parseArgs(args); @@ -133,19 +146,6 @@ public class ClientCompatibilityTest { System.exit(0); } - private static byte[] asByteArray(long a) { - ByteBuffer buf = ByteBuffer.allocate(8); - buf.putLong(a); - return buf.array(); - } - - private static byte[] asByteArray(long a, long b) { - ByteBuffer buf = ByteBuffer.allocate(16); - buf.putLong(a); - buf.putLong(b); - return buf.array(); - } - private static String toHexString(byte[] buf) { StringBuilder bld = new StringBuilder(); for (byte b : buf) { @@ -169,8 +169,16 @@ public class ClientCompatibilityTest { ClientCompatibilityTest(TestConfig testConfig) { this.testConfig = testConfig; long curTime = Time.SYSTEM.milliseconds(); - this.message1 = asByteArray(curTime); - this.message2 = asByteArray(curTime, curTime); + + ByteBuffer buf = ByteBuffer.allocate(8); + buf.putLong(curTime); + this.message1 = buf.array(); + + ByteBuffer buf2 = ByteBuffer.allocate(4096); + for (long i = 0; i < buf2.capacity(); i += 8) { + buf2.putLong(curTime + i); + } + this.message2 = buf2.array(); } void run() throws Exception { @@ -242,6 +250,7 @@ public class ClientCompatibilityTest { public void testConsume(final long prodTimeMs) throws Exception { Properties consumerProps = new Properties(); consumerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, testConfig.bootstrapServer); + consumerProps.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 512); ClientCompatibilityTestDeserializer deserializer = new ClientCompatibilityTestDeserializer(testConfig.expectClusterId); final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps, deserializer, deserializer); @@ -318,7 +327,6 @@ public class ClientCompatibilityTest { throw new UnsupportedOperationException(); } }; - byte[] next = iter.next(); try { compareArrays(message1, next); @@ -327,13 +335,23 @@ public class ClientCompatibilityTest { throw new RuntimeException("The first message in this topic was not ours. Please use a new topic when " + "running this program."); } - next = iter.next(); try { - compareArrays(message2, next); - log.debug("Found second message..."); - } catch (RuntimeException e) { - throw new RuntimeException("The second message in this topic was not ours. Please use a new topic when " + - "running this program."); + next = iter.next(); + if (testConfig.expectRecordTooLargeException) + throw new RuntimeException("Expected to get a RecordTooLargeException when reading a record " + + "bigger than " + ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG); + try { + compareArrays(message2, next); + } catch (RuntimeException e) { + System.out.println("The second message in this topic was not ours. Please use a new " + + "topic when running this program."); + System.exit(1); + } + } catch (RecordTooLargeException e) { + log.debug("Got RecordTooLargeException", e); + if (!testConfig.expectRecordTooLargeException) + throw new RuntimeException("Got an unexpected RecordTooLargeException when reading a record " + + "bigger than " + ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG); } log.debug("Closing consumer."); consumer.close();
