[kafka] branch trunk updated: MINOR: improve stabilty of ProcessorStateManagerTest (#6240)

[gu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

guozhang pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 2fb9b03  MINOR: improve stabilty of ProcessorStateManagerTest (#6240)
2fb9b03 is described below

commit 2fb9b03d6f19abada6babeec777a4452e989bc15
Author: Matthias J. Sax <mj...@apache.org>
AuthorDate: Fri Feb 8 13:47:53 2019 -0800

    MINOR: improve stabilty of ProcessorStateManagerTest (#6240)
    
    This PR addressed the following test failure:
    
    ```
    java.lang.AssertionError:
    Expected: a string starting with "process-state-manager-test Failed to write offset checkpoint file to ["
    but: was "[AdminClient clientId=adminclient-874] Connection to node -1 (localhost/127.0.0.1:8080) could not be established. Broker may not be available."
    ```
    
    Reviewers: Bill Bejeck <bi...@confluent.io>, John Roesler <jo...@confluent.io>, Guozhang Wang <gu...@confluent.io>
---
 .../internals/ProcessorStateManagerTest.java         | 20 +++++++++++---------
 1 file changed, 11 insertions(+), 9 deletions(-)

diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
index ce76749..119790f 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
@@ -53,9 +53,7 @@ import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.util.Collections.singletonList;
-import static org.hamcrest.CoreMatchers.endsWith;
 import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.startsWith;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
 import static org.junit.Assert.assertEquals;
@@ -619,14 +617,18 @@ public class ProcessorStateManagerTest {
         stateMgr.checkpoint(Collections.singletonMap(persistentStorePartition, 10L));
         LogCaptureAppender.unregister(appender);
 
-        final List<LogCaptureAppender.Event> messages = appender.getEvents();
+        boolean foundExpectedLogMessage = false;
+        for (final LogCaptureAppender.Event event : appender.getEvents()) {
+            if ("WARN".equals(event.getLevel())
+                && event.getMessage().startsWith("process-state-manager-test Failed to write offset checkpoint file to [")
+                && event.getMessage().endsWith(".checkpoint]")
+                && event.getThrowableInfo().get().startsWith("java.io.FileNotFoundException: ")) {
 
-        final LogCaptureAppender.Event lastEvent = messages.get(messages.size() - 1);
-
-        assertThat(lastEvent.getLevel(), is("WARN"));
-        assertThat(lastEvent.getMessage(), startsWith("process-state-manager-test Failed to write offset checkpoint file to ["));
-        assertThat(lastEvent.getMessage(), endsWith(".checkpoint]"));
-        assertThat(lastEvent.getThrowableInfo().get(), startsWith("java.io.FileNotFoundException: "));
+                foundExpectedLogMessage = true;
+                break;
+            }
+        }
+        assertTrue(foundExpectedLogMessage);
     }
 
     @Test