From 0c413b80503d6e4a3695692f07bbe08b4f358e8d Mon Sep 17 00:00:00 2001 From: "massinissa.ikhelef@gmail.com" <7047076+mikhelef@users.noreply.github.com> Date: Fri, 1 Nov 2024 10:20:23 +0100 Subject: [PATCH 1/2] Close PC when InvalidPidMappingException Add fix in CHANGELOG.adoc Add fix in CHANGELOG.adoc add failure reason when closing due to invalidPidMappingException update UT with failurecause make failure reason setter protected refactoring refactoring --- CHANGELOG.adoc | 1 + .../ParallelEoSStreamProcessor.java | 6 ++- .../AbstractParallelEoSStreamProcessor.java | 11 +++++ .../ParallelEoSStreamProcessorTest.java | 48 +++++++++++++++++++ 4 files changed, 65 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.adoc b/CHANGELOG.adoc index 97d120ef42..4bb1344a68 100644 --- a/CHANGELOG.adoc +++ b/CHANGELOG.adoc @@ -17,6 +17,7 @@ endif::[] === Fixes +* fix: close parallel consumer on transactional mode when InvalidPidMappingException (#830) * fix: include inflight message count in polling backpressure logic (#836) * fix: message loss on closing or partitions revoked (#827) fixes (#826) * fix: unbounded retry queue growth preventing polling from being throttled and leading to OOM (#834) fixes (#832, #817) diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessor.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessor.java index 8df85f171e..4743923778 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessor.java +++ b/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> 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); } diff --git a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java b/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java index 2918717eb0..8b7a687867 100644 --- a/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java +++ b/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; diff --git a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java b/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java index e9b9b09e7e..4726a38617 100644 --- a/parallel-consumer-core/src/test/java/io/confluent/parallelconsumer/ParallelEoSStreamProcessorTest.java +++ b/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 pcSpy = spy(parallelConsumer); + final InvalidPidMappingException invalidPidMappingException = new InvalidPidMappingException("InvalidPidMappingException exception"); + + // use mocked producer manager + ProducerManager 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 From 9d691d120969a3db7d5b77c64dbc4df5b79b14fd Mon Sep 17 00:00:00 2001 From: "massinissa.ikhelef@gmail.com" <7047076+mikhelef@users.noreply.github.com> Date: Fri, 1 Nov 2024 16:42:32 +0100 Subject: [PATCH 2/2] Add == 0.5.3.3 in CHANGELOG.adoc and update README.adoc --- CHANGELOG.adoc | 8 +++++++- README.adoc | 7 +++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.adoc b/CHANGELOG.adoc index 4bb1344a68..d59068d799 100644 --- a/CHANGELOG.adoc +++ b/CHANGELOG.adoc @@ -13,11 +13,17 @@ NOTE:: Dependency version bumps are not listed here. ifndef::github_name[] toc::[] endif::[] -== 0.5.3.2 + +== 0.5.3.3 === Fixes * fix: close parallel consumer on transactional mode when InvalidPidMappingException (#830) + +== 0.5.3.2 + +=== Fixes + * fix: include inflight message count in polling backpressure logic (#836) * fix: message loss on closing or partitions revoked (#827) fixes (#826) * fix: unbounded retry queue growth preventing polling from being throttled and leading to OOM (#834) fixes (#832, #817) diff --git a/README.adoc b/README.adoc index d8d542cc1c..0e1dbd1e5b 100644 --- a/README.adoc +++ b/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