Improve test coverage of StreamsKafkaClient
The commit brings improved test coverage for StreamsKafkaClientTest.java
Author: Andrey Dyachkov
Reviewers: Damian Guy
Closes from adyach/kafka-4643
@@ -18,17 +18,23 @@ package org.apache.kafka.streams.processor.internals;
import org.apache.kafka.clients.MockClient;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.metrics.MetricsReporter;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.ApiError;
+import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.requests.CreateTopicsRequest;
import org.apache.kafka.common.requests.CreateTopicsResponse;
import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.errors.StreamsException;
import org.junit.Before;
import org.junit.Test;
@@ -124,6 +130,48 @@ public class StreamsKafkaClientTest {
verifyCorrectTopicConfigs(streamsKafkaClient, topicConfigWithNoOverrides, Collections.singletonMap("cleanup.policy", "delete"));
}
+ @Test(expected = StreamsException.class)
+ public void shouldThrowStreamsExceptionOnEmptyBrokerCompatibilityResponse() {
+ kafkaClient.prepareResponse(null);
+ final StreamsKafkaClient streamsKafkaClient = createStreamsKafkaClient();
+ streamsKafkaClient.checkBrokerCompatibility(false);
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldThrowStreamsExceptionWhenBrokerCompatibilityResponseInconsistent() {
+ kafkaClient.prepareResponse(new ProduceResponse(Collections.<TopicPartition, ProduceResponse.PartitionResponse>emptyMap()));
+ final StreamsKafkaClient streamsKafkaClient = createStreamsKafkaClient();
+ streamsKafkaClient.checkBrokerCompatibility(false);
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldRequireBrokerVersion0101OrHigherWhenEosDisabled() {
+ kafkaClient.prepareResponse(new ApiVersionsResponse(Errors.NONE, Collections.singletonList(new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE))));
+ final StreamsKafkaClient streamsKafkaClient = createStreamsKafkaClient();
+ streamsKafkaClient.checkBrokerCompatibility(false);
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldRequireBrokerVersions0110OrHigherWhenEosEnabled() {
+ kafkaClient.prepareResponse(new ApiVersionsResponse(Errors.NONE, Collections.singletonList(new ApiVersionsResponse.ApiVersion(ApiKeys.CREATE_TOPICS))));
+ final StreamsKafkaClient streamsKafkaClient = createStreamsKafkaClient();
+ streamsKafkaClient.checkBrokerCompatibility(true);
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldThrowStreamsExceptionOnEmptyFetchMetadataResponse() {
+ kafkaClient.prepareResponse(null);
+ final StreamsKafkaClient streamsKafkaClient = createStreamsKafkaClient();
+ streamsKafkaClient.fetchMetadata();
+ }
+
+ @Test(expected = StreamsException.class)
+ public void shouldThrowStreamsExceptionWhenFetchMetadataResponseInconsistent() {
+ kafkaClient.prepareResponse(new ProduceResponse(Collections.<TopicPartition, ProduceResponse.PartitionResponse>emptyMap()));
+ final StreamsKafkaClient streamsKafkaClient = createStreamsKafkaClient();
+ streamsKafkaClient.fetchMetadata();
+ }
+
private void verifyCorrectTopicConfigs(final StreamsKafkaClient streamsKafkaClient,
final InternalTopicConfig internalTopicConfig,
final Map<String, String> expectedConfigs) {
