[GitHub] [kafka] philipnee commented on a diff in pull request #13303: KAFKA-14761 Adding integration test for the prototype consumer

[via GitHub]

philipnee commented on code in PR #13303:
URL: https://github.com/apache/kafka/pull/13303#discussion_r1127257845


##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -16,90 +16,77 @@
  */
 package org.apache.kafka.clients.consumer.internals;
 
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.OffsetResetStrategy;
-import org.apache.kafka.clients.consumer.internals.events.EventHandler;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.internals.ClusterResourceListeners;
-import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Optional;
 
-import static java.util.Collections.singleton;
-import static 
org.apache.kafka.clients.consumer.ConsumerConfig.CLIENT_ID_CONFIG;
+import static 
org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG;
 import static 
org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG;
 import static 
org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG;
-import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class PrototypeAsyncConsumerTest {
-    private Map<String, Object> properties;
-    private SubscriptionState subscriptionState;
-    private MockTime time;
-    private LogContext logContext;
-    private Metrics metrics;
-    private ClusterResourceListeners clusterResourceListeners;
-    private Optional<String> groupId;
-    private String clientId;
-    private EventHandler eventHandler;
+
+    private Consumer<?, ?> consumer;
+    private Map<String, Object> consumerProps = new HashMap<>();
+
+    private final Time time = new MockTime();
 
     @BeforeEach
     public void setup() {
-        this.subscriptionState = Mockito.mock(SubscriptionState.class);
-        this.eventHandler = Mockito.mock(DefaultEventHandler.class);
-        this.logContext = new LogContext();
-        this.time = new MockTime();
-        this.metrics = new Metrics(time);
-        this.groupId = Optional.empty();
-        this.clientId = "client-1";
-        this.clusterResourceListeners = new ClusterResourceListeners();
-        this.properties = new HashMap<>();
-        this.properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
-                "localhost" +
-                ":9999");
-        this.properties.put(KEY_DESERIALIZER_CLASS_CONFIG, 
StringDeserializer.class);
-        this.properties.put(VALUE_DESERIALIZER_CLASS_CONFIG, 
StringDeserializer.class);
-        this.properties.put(CLIENT_ID_CONFIG, "test-client");
+        injectConsumerConfigs();
+    }
+
+    @AfterEach
+    public void cleanup() {
+        if (consumer != null) {
+            consumer.close(Duration.ZERO);
+        }
     }
+
     @Test
-    public void testSubscription() {
-        this.subscriptionState =
-                new SubscriptionState(new LogContext(), 
OffsetResetStrategy.EARLIEST);
-        PrototypeAsyncConsumer<String, String> consumer =
-                setupConsumerWithDefault();
-        subscriptionState.subscribe(singleton("t1"),
-                new NoOpConsumerRebalanceListener());
-        assertEquals(1, consumer.subscription().size());
+    public void testBackgroundThreadRunning() {
+        consumer = newConsumer(time, new StringDeserializer(), new 
StringDeserializer());
     }
 
     @Test
     public void testUnimplementedException() {
-        PrototypeAsyncConsumer<String, String> consumer =
-                setupConsumerWithDefault();
+        consumer = newConsumer(time, new StringDeserializer(), new 
StringDeserializer());
         assertThrows(KafkaException.class, consumer::assignment, "not 
implemented exception");
     }
 
-    public PrototypeAsyncConsumer<String, String> setupConsumerWithDefault() {
-        ConsumerConfig config = new ConsumerConfig(properties);
-        return new PrototypeAsyncConsumer<>(
-                this.time,
-                this.logContext,
-                config,
-                this.subscriptionState,
-                this.eventHandler,
-                this.metrics,
-                this.clusterResourceListeners,
-                this.groupId,
-                this.clientId,
-                0);
+    private ConsumerMetadata createMetadata(SubscriptionState subscription) {
+        return new ConsumerMetadata(0, Long.MAX_VALUE, false, false,
+                subscription, new LogContext(), new 
ClusterResourceListeners());
+    }
+
+    private void injectConsumerConfigs() {
+        consumerProps.put(BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");

Review Comment:
   I think that's the intent: I don't think it is necessary to make the actual 
connection to the broker on the level of unit test.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org