minor: fixes #486: Missing generics in JStreamParallelStreamProcessor

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/JStreamParallelStreamProcessor.java

@@ -14,7 +14,7 @@
 
 public interface JStreamParallelStreamProcessor<K, V> extends DrainingCloseable {
 
-    static JStreamParallelStreamProcessor createJStreamEosStreamProcessor(ParallelConsumerOptions<?, ?> options) {
+    static <K, V> JStreamParallelStreamProcessor<K, V> createJStreamEosStreamProcessor(ParallelConsumerOptions<K, V> options) {
         return new JStreamParallelEoSStreamProcessor<>(options);
     }
 

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumer.java

@@ -27,6 +27,11 @@
 // end::javadoc[]
 public interface ParallelConsumer<K, V> extends DrainingCloseable {
 
+    /**
+     * @return true if the system has either closed, or has crashed
+     */
+    boolean isClosedOrFailed();
+
     /**
      * @see KafkaConsumer#subscribe(Collection)
      */

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelConsumerOptions.java

@@ -24,6 +24,16 @@
 
 /**
  * The options for the {@link AbstractParallelEoSStreamProcessor} system.
+ * <p>
+ * The important options to look at are:
+ * <p>
+ * {@link #ordering}, {@link #maxConcurrency} and {@link #batchSize}.
+ * <p>
+ * If you want to go deeper, look at {@link #defaultMessageRetryDelay}, {@link #retryDelayProvider} and
+ * {@link #commitMode}.
+ * <p>
+ * Note: The only required option is the {@link #consumer} ({@link #producer} is only needed if you use the Produce
+ * flows). All other options have sensible defaults.
  *
  * @author Antony Stubbs
  * @see #builder()

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java

@@ -186,6 +186,7 @@ public static <K, V> ControllerEventMessage<K, V> of(WorkContainer<K, V> work) {
      */
     private Instant lastCommitTime;
 
+    @Override
     public boolean isClosedOrFailed() {
         boolean closed = state == State.closed;
         boolean doneOrCancelled = false;

parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerOffsetCommitter.java

@@ -102,9 +102,8 @@ protected void commitOffsets(final Map<TopicPartition, OffsetAndMetadata> offset
                     }
                 });
             }
-            default -> {
-                throw new IllegalArgumentException("Cannot use " + commitMode + " when using " + this.getClass().getSimpleName());
-            }
+            default ->
+                    throw new IllegalArgumentException("Cannot use " + commitMode + " when using " + this.getClass().getSimpleName());
         }
     }
 

parallel-consumer-core/src/test-integration/java/io/confluent/parallelconsumer/integrationTests/BrokerIntegrationTest.java

@@ -55,7 +55,7 @@ public abstract class BrokerIntegrationTest<K, V> {
     public static KafkaContainer kafkaContainer = createKafkaContainer(null);
 
     public static KafkaContainer createKafkaContainer(String logSegmentSize) {
-        KafkaContainer base = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.2.2"))
+        KafkaContainer base = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.3.0"))
                 .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1") //transaction.state.log.replication.factor
                 .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1") //transaction.state.log.min.isr
                 .withEnv("KAFKA_TRANSACTION_STATE_LOG_NUM_PARTITIONS", "1") //transaction.state.log.num.partitions