fixes #830: Close PC when InvalidPidMappingException

CHANGELOG.adoc

@@ -13,6 +13,13 @@ NOTE:: Dependency version bumps are not listed here.
 ifndef::github_name[]
 toc::[]
 endif::[]
+
+== 0.5.3.3
+
+=== Fixes
+
+* fix: close parallel consumer on transactional mode when InvalidPidMappingException (#830)
+
 == 0.5.3.2
 
 === Fixes

README.adoc

@@ -1533,6 +1533,13 @@ NOTE:: Dependency version bumps are not listed here.
 ifndef::github_name[]
 toc::[]
 endif::[]
+
+== 0.5.3.3
+
+=== Fixes
+
+* fix: close parallel consumer on transactional mode when InvalidPidMappingException (#830)
+
 == 0.5.3.2
 
 === Fixes

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

@@ -1,7 +1,7 @@
 package io.confluent.parallelconsumer;
 
 /*-
- * Copyright (C) 2020-2023 Confluent, Inc.
+ * Copyright (C) 2020-2024 Confluent, Inc.
  */
 
 import io.confluent.csid.utils.TimeUtils;
@@ -13,6 +13,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.errors.InvalidPidMappingException;
 import pl.tlinkowski.unij.api.UniLists;
 
 import java.util.ArrayList;
@@ -132,6 +133,9 @@ private List<ConsumeProduceResult<K, V, K, V>> processAndProduceResults(final Fu
                 }
                 return null; // return from timer function
             });
+        } catch (InvalidPidMappingException invalidPidMappingException) {
+            log.error("Closing parallel Consumer due to InvalidPidMappingException", invalidPidMappingException);
+            this.closeOnException(invalidPidMappingException);
         } catch (Exception e) {
             throw new InternalRuntimeException("Error while waiting for produce results", e);
         }

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

@@ -568,6 +568,17 @@ public void close() {
         closeDontDrainFirst();
     }
 
+    /**
+     * Close the system without draining and set failure reason
+     * @param exception
+     *
+     * @see State#DRAINING
+     */
+    public void closeOnException(Exception exception){
+        this.failureReason = exception;
+        closeDontDrainFirst();
+    }
+
     @Override
     public void close(Duration timeout, DrainingMode drainMode) {
         shutdownTimeout = timeout;

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

@@ -7,12 +7,16 @@
 import io.confluent.csid.utils.*;
 import io.confluent.parallelconsumer.ParallelConsumerOptions.CommitMode;
 import io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor;
+import io.confluent.parallelconsumer.internal.ProducerManager;
+import io.confluent.parallelconsumer.internal.State;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.errors.InvalidPidMappingException;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serdes;
+import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Assumptions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Disabled;
@@ -22,7 +26,9 @@
 import org.junit.jupiter.params.provider.EnumSource;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
 
+import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
@@ -86,6 +92,47 @@ public void failingActionNothingCommitted(CommitMode commitMode) {
         assertCommits(of(), "All erroring, so nothing committed except initial");
     }
 
+    @Test
+    @SneakyThrows
+    public void closePCWhenInvalidPidMappingException() {
+        setupParallelConsumerInstance(PERIODIC_CONSUMER_ASYNCHRONOUS);
+
+        MockitoAnnotations.openMocks(this);
+        final ParallelEoSStreamProcessor<String, String> pcSpy = spy(parallelConsumer);
+        final InvalidPidMappingException invalidPidMappingException = new InvalidPidMappingException("InvalidPidMappingException exception");
+
+        // use mocked producer manager
+        ProducerManager<String, String> producerManager = mock(ProducerManager.class);
+        Field producerManagerField = AbstractParallelEoSStreamProcessor.class.getDeclaredField("producerManager");
+        producerManagerField.setAccessible(true);
+        producerManagerField.set(pcSpy, Optional.of(producerManager));
+
+        when(producerManager.beginProducing(any())).thenReturn(mock(ProducerManager.ProducingLock.class));
+        when(producerManager.produceMessages(any())).thenThrow(invalidPidMappingException);
+
+        CountDownLatch latch = new CountDownLatch(1);
+        // count down latch if close is invoked
+        Mockito.doAnswer(invocation -> {
+            Object result = invocation.callRealMethod();
+            latch.countDown();
+            return result;
+        }).when(pcSpy).close(any());
+
+        pcSpy.pollAndProduceMany((record) -> of(new ProducerRecord<>("outputTopic", record.key(), record.value())));
+
+        latch.await();
+
+        //check failure cause
+        Assertions.assertThat(pcSpy.getFailureCause().equals(invalidPidMappingException)).isTrue();
+
+        //check state is CLOSED
+        Field state = AbstractParallelEoSStreamProcessor.class.getDeclaredField("state");
+        state.setAccessible(true);
+        Assertions.assertThat(State.CLOSED.equals(state.get(pcSpy))).isTrue();
+
+    }
+
+
     @ParameterizedTest()
     @EnumSource(CommitMode.class)
     @SneakyThrows
@@ -115,6 +162,7 @@ public void executorThreadsInterruptedOnShutdownTimeout(CommitMode commitMode) {
         assertThat(interrupted).isTrue();
     }
 
+
     @ParameterizedTest()
     @EnumSource(CommitMode.class)
     @SneakyThrows