From 67820c17aba5a212ba00ef33c36de2540cdb303e Mon Sep 17 00:00:00 2001 From: Jim Hahn Date: Thu, 1 Jul 2021 11:41:00 -0400 Subject: Wait after fetch exception on topic When dmaap is inaccessible for some reason, the topic source frequently enters a fast fail loop, rapidly filling up the log. Modified the code to wait the configured fetchTimeout when this occurs. With any luck, this will also fix the sporadic kubernetes crash-fail loops sometimes seen with the xacml-pdp pod. Modified to limit how long it will sleep after a failure, regardless of the fetchTimeout that was specified. Issue-ID: POLICY-3457 Change-Id: I88e360fb1d31197b46f4959e5ea0ea2d741ad25c Signed-off-by: Jim Hahn --- .../event/comm/bus/internal/BusConsumerTest.java | 40 ++++++++++++++++++---- 1 file changed, 34 insertions(+), 6 deletions(-) (limited to 'policy-endpoints/src/test') diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java index aba05887..21050f97 100644 --- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java +++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java @@ -46,6 +46,7 @@ import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.Dmaa import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapConsumerWrapper; import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapDmeConsumerWrapper; import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.FetchingBusConsumer; +import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties; import org.powermock.reflect.Whitebox; public class BusConsumerTest extends TopicTestBase { @@ -59,10 +60,30 @@ public class BusConsumerTest extends TopicTestBase { super.setUp(); } + @Test + public void testFetchingBusConsumer() throws InterruptedException { + // should not be negative + var cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(-1).build()); + assertThat(cons.getSleepTime()).isEqualTo(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH); + + // should not be zero + cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(0).build()); + assertThat(cons.getSleepTime()).isEqualTo(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH); + + // should not be too large + cons = new FetchingBusConsumerImpl( + makeBuilder().fetchTimeout(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH + 100).build()); + assertThat(cons.getSleepTime()).isEqualTo(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH); + + // should not be what was specified + cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(100).build()); + assertThat(cons.getSleepTime()).isEqualTo(100); + } + @Test public void testFetchingBusConsumerSleepAfterFetchFailure() throws InterruptedException { - var cons = new FetchingBusConsumer(makeBuilder().fetchTimeout(SHORT_TIMEOUT_MILLIS).build()) { + var cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(SHORT_TIMEOUT_MILLIS).build()) { private CountDownLatch started = new CountDownLatch(1); @@ -71,11 +92,6 @@ public class BusConsumerTest extends TopicTestBase { started.countDown(); super.sleepAfterFetchFailure(); } - - @Override - public Iterable fetch() throws IOException { - return null; - } }; // full sleep @@ -278,4 +294,16 @@ public class BusConsumerTest extends TopicTestBase { public void testDmaapDmeConsumerWrapper_InvalidPartner() throws Exception { new DmaapDmeConsumerWrapper(makeBuilder().partner(null).build()); } + + private static class FetchingBusConsumerImpl extends FetchingBusConsumer { + + protected FetchingBusConsumerImpl(BusTopicParams busTopicParams) { + super(busTopicParams); + } + + @Override + public Iterable fetch() throws IOException { + return null; + } + } } -- cgit 1.2.3-korg