Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1331: KMLC - treat auth. exceptions as fatal

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/1331

Authorization errors create a log storm; stop the container for
such exceptions in the same way we do for `NoOffsetForPartitionException`.

**cherry-pick to 2.2.x**
This commit is contained in:
Gary Russell
2019-12-13 09:55:09 -05:00
committed by Artem Bilan
parent ceb8a99dc9
commit 09e775f098
2 changed files with 37 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -53,6 +53,7 @@ import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.WakeupException;
import org.springframework.context.ApplicationContext;
@@ -908,6 +909,11 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
ListenerConsumer.this.logger.error(nofpe, "No offset and no reset policy");
break;
}
catch (AuthorizationException ae) {
this.fatalError = true;
ListenerConsumer.this.logger.error(ae, "Authorization Exception");
break;
}
catch (Exception e) {
handleConsumerException(e);
}
@@ -1091,7 +1097,7 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
}
}
else {
this.logger.error("No offset and no reset policy; stopping container");
this.logger.error("Fatal consumer exception; stopping container");
KafkaMessageListenerContainer.this.stop();
}
this.monitorTask.cancel(true);

30
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java
View File

@@ -48,6 +48,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
@@ -149,6 +150,35 @@ public class ConcurrentMessageListenerContainerMockTests {
container.stop();
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
void testConsumerExitWhenNotAuthorized() throws InterruptedException {
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
final Consumer consumer = mock(Consumer.class);
willAnswer(invocation -> {
Thread.sleep(100);
throw new GroupAuthorizationException("grp");
}).given(consumer).poll(any());
CountDownLatch latch = new CountDownLatch(1);
willAnswer(invocation -> {
latch.countDown();
return null;
}).given(consumer).close();
given(consumerFactory.createConsumer("grp", "", "-0", KafkaTestUtils.defaultPropertyOverrides()))
.willReturn(consumer);
ContainerProperties containerProperties = new ContainerProperties("foo");
containerProperties.setGroupId("grp");
containerProperties.setMessageListener((MessageListener) record -> { });
containerProperties.setMissingTopicsFatal(false);
containerProperties.setShutdownTimeout(10);
ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(consumerFactory,
containerProperties);
container.start();
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
verify(consumer).close();
container.stop();
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
@DisplayName("Seek on TL callback when idle")