rebase compile update

Author: astubbs

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

@@ -33,7 +33,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import static io.confluent.csid.utils.LatchTestUtils.awaitLatch;
 import static io.confluent.csid.utils.StringUtils.msg;
@@ -345,7 +344,7 @@ public List<Integer> getCommitHistoryFlattened() {
     public List<OffsetAndMetadata> getCommitHistoryFlattenedMeta() {
         return (isUsingTransactionalProducer())
                 ? ktu.getProducerCommitsMeta(producerSpy)
-                : extractAllPartitionsOffsetsSequentiallyMeta();
+                : extractAllPartitionsOffsetsSequentiallyMeta(true);
     }
 
     public void assertCommits(List<Integer> offsets, String description) {
@@ -381,26 +380,26 @@ public void assertCommits(List<Integer> offsets, Optional<String> description) {
      * Flattens the offsets of all partitions into a single sequential list
      */
     protected List<Integer> extractAllPartitionsOffsetsSequentially(boolean trimGenesis) {
-        return extractAllPartitionsOffsetsSequentiallyMeta().stream().
+        return extractAllPartitionsOffsetsSequentiallyMeta(trimGenesis).stream().
                 map(x -> (int) x.offset()) // int cast a luxury in test context - no big offsets
                 .collect(Collectors.toList());
     }
 
     /**
      * Flattens the offsets of all partitions into a single sequential list
      */
-    protected List<OffsetAndMetadata> extractAllPartitionsOffsetsSequentiallyMeta() {
+    protected List<OffsetAndMetadata> extractAllPartitionsOffsetsSequentiallyMeta(boolean trimGenesis) {
         // copy the list for safe concurrent access
         List<Map<TopicPartition, OffsetAndMetadata>> history = new ArrayList<>(consumerSpy.getCommitHistoryInt());
         return history.stream()
                 .flatMap(commits ->
                         {
-                            Collection<OffsetAndMetadata> values = new ArrayList<>(commits.values()); // 4 debugging
-                            Stream<Integer> rawOffsets = values.stream().map(meta -> (int) meta.offset());
+                            var rawValues = new ArrayList<>(commits.values()).stream(); // 4 debugging
+//                            Stream<Integer> rawOffsets = values.stream().map(meta -> (int) meta.offset());
                             if (trimGenesis)
-                                return rawOffsets.filter(x -> x != 0);
+                                return rawValues.filter(x -> x.offset() != 0);
                             else
-                                return rawOffsets; // int cast a luxury in test context - no big offsets
+                                return rawValues; // int cast a luxury in test context - no big offsets
                         }
                 ).collect(Collectors.toList());
     }

parallel-consumer-core/src/test/resources/logback-test.xml

@@ -28,7 +28,7 @@
 
     <!-- primary -->
     <logger name="io.confluent.parallelconsumer" level="info"/>
-        <logger name="io.confluent.parallelconsumer" level="debug"/>
+<!--        <logger name="io.confluent.parallelconsumer" level="debug"/>-->
 
     <!--    <logger name="io.confluent.csid" level="debug"/>-->