review

Author: astubbs

parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/MockConsumerTest.java

@@ -1,5 +1,9 @@
 package io.confluent.parallelconsumer;
 
+/*-
+ * Copyright (C) 2020-2022 Confluent, Inc.
+ */
+
 import io.confluent.csid.utils.LongPollingMockConsumer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.MockConsumer;
@@ -26,7 +30,7 @@
  * @see LongPollingMockConsumer#revokeAssignment
  */
 @Slf4j
-class MockConsumerTest {//} extends ParallelEoSStreamProcessorTestBase {
+class MockConsumerTest {
 
     private final String topic = MockConsumerTest.class.getSimpleName();
 

parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java

@@ -823,15 +823,15 @@ void optionsGroupIdRequiredAndAutoCommitDisabled() {
         optionsBuilder.consumer(new KafkaConsumer<>(properties, deserializer, deserializer));
         assertThat(catchThrowable(() -> parallelConsumer = initPollingAsyncConsumer(optionsBuilder.build())))
                 .as("Should error on auto commit enabled by default")
-                .isInstanceOf(IllegalArgumentException.class)
+                .isInstanceOf(ParallelConsumerException.class)
                 .hasMessageContainingAll("auto", "commit", "disabled");
 
         // fail auto commit disabled
         properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
         optionsBuilder.consumer(new KafkaConsumer<>(properties, deserializer, deserializer));
         assertThat(catchThrowable(() -> parallelConsumer = initPollingAsyncConsumer(optionsBuilder.build())))
                 .as("Should error on auto commit enabled")
-                .isInstanceOf(IllegalArgumentException.class)
+                .isInstanceOf(ParallelConsumerException.class)
                 .hasMessageContainingAll("auto", "commit", "disabled");
 
         // set missing auto commit