[
https://issues.apache.org/jira/browse/KAFKA-7195?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16554004#comment-16554004
]
ASF GitHub Bot commented on KAFKA-7195:
---------------------------------------
rajinisivaram closed pull request #5418: KAFKA-7195: Fix
StreamStreamJoinIntegrationTest test failures
URL: https://github.com/apache/kafka/pull/5418
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java
b/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java
index 80ab60647ad..3e29fc2a29b 100644
---
a/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/integration/AbstractJoinIntegrationTest.java
@@ -163,7 +163,7 @@ void prepareEnvironment() throws InterruptedException {
@After
public void cleanup() throws InterruptedException {
- CLUSTER.deleteTopicsAndWait(120000, INPUT_TOPIC_LEFT,
INPUT_TOPIC_RIGHT, OUTPUT_TOPIC);
+ CLUSTER.deleteAllTopicsAndWait(120000);
}
private void checkResult(final String outputTopic, final List<String>
expectedResult) throws InterruptedException {
diff --git
a/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
b/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
index ce6324df617..ab52649dee4 100644
---
a/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
+++
b/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java
@@ -274,6 +274,24 @@ public void deleteTopicsAndWait(final long timeoutMs,
final String... topics) th
}
}
+ /**
+ * Deletes all topics and blocks until all topics got deleted.
+ *
+ * @param timeoutMs the max time to wait for the topics to be deleted
(does not block if {@code <= 0})
+ */
+ public void deleteAllTopicsAndWait(final long timeoutMs) throws
InterruptedException {
+ final Set<String> topics = new
HashSet<>(JavaConverters.seqAsJavaListConverter(zkUtils.getAllTopics()).asJava());
+ for (final String topic : topics) {
+ try {
+ brokers[0].deleteTopic(topic);
+ } catch (final UnknownTopicOrPartitionException e) { }
+ }
+
+ if (timeoutMs > 0) {
+ TestUtils.waitForCondition(new TopicsDeletedCondition(topics),
timeoutMs, "Topics not deleted after " + timeoutMs + " milli seconds.");
+ }
+ }
+
public void deleteAndRecreateTopics(final String... topics) throws
InterruptedException {
deleteTopicsAndWait(TOPIC_DELETION_TIMEOUT, topics);
createTopics(topics);
@@ -295,6 +313,10 @@ private TopicsDeletedCondition(final String... topics) {
Collections.addAll(deletedTopics, topics);
}
+ public TopicsDeletedCondition(final Set<String> topics) {
+ deletedTopics.addAll(topics);
+ }
+
@Override
public boolean conditionMet() {
final Set<String> allTopics = new HashSet<>(
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> StreamStreamJoinIntegrationTest fails in 2.0 Jenkins
> ----------------------------------------------------
>
> Key: KAFKA-7195
> URL: https://issues.apache.org/jira/browse/KAFKA-7195
> Project: Kafka
> Issue Type: Test
> Reporter: Ted Yu
> Assignee: Manikumar
> Priority: Major
>
> From
> https://builds.apache.org/job/kafka-2.0-jdk8/87/testReport/junit/org.apache.kafka.streams.integration/StreamStreamJoinIntegrationTest/testOuter_caching_enabled___false_/
> :
> {code}
> java.lang.AssertionError:
> Expected: is <[A-null]>
> but: was <[A-a, A-b, A-c, A-d]>
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:8)
> at
> org.apache.kafka.streams.integration.AbstractJoinIntegrationTest.checkResult(AbstractJoinIntegrationTest.java:171)
> at
> org.apache.kafka.streams.integration.AbstractJoinIntegrationTest.runTest(AbstractJoinIntegrationTest.java:212)
> at
> org.apache.kafka.streams.integration.AbstractJoinIntegrationTest.runTest(AbstractJoinIntegrationTest.java:184)
> at
> org.apache.kafka.streams.integration.StreamStreamJoinIntegrationTest.testOuter(StreamStreamJoinIntegrationTest.java:198)
> {code}
> However, some test output was missing:
> {code}
> [2018-07-23 20:51:36,363] INFO Socket c
> ...[truncated 1627692 chars]...
> 671)
> {code}
> I ran the test locally which passed.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
