root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: Fix KStreamKTableJoinTest and StreamTaskTest (#9357) 

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>
This commit is contained in:
leah 2020-09-30 22:07:23 -05:00 committed by GitHub
parent a15387f34d
commit 95986a8f48
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 21 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
checkstyle/suppressions.xml
View File

@ -208,6 +208,9 @@
<suppress checks="MethodLength"
files="KStreamSlidingWindowAggregateTest.java"/>
<suppress checks="ClassFanOutComplexity"
files="StreamTaskTest.java"/>
<!-- Streams test-utils -->
<suppress checks="ClassFanOutComplexity"
files="TopologyTestDriver.java"/>

4
streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinTest.java
View File

@ -255,7 +255,7 @@ public class KStreamKTableJoinTest {
}
assertThat(
appender.getMessages(),
hasItem("Skipping record due to null key or value. key=[null] value=[A] topic=[streamTopic] partition=[0] "
hasItem("Skipping record due to null join key or value. key=[null] value=[A] topic=[streamTopic] partition=[0] "
+ "offset=[0]"));
}
}
@ -282,7 +282,7 @@ public class KStreamKTableJoinTest {
assertThat(
appender.getMessages(),
hasItem("Skipping record due to null key or value. key=[1] value=[null] topic=[streamTopic] partition=[0] "
hasItem("Skipping record due to null join key or value. key=[1] value=[null] topic=[streamTopic] partition=[0] "
+ "offset=[0]")
);
}

32
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamTaskTest.java
View File

@ -2050,22 +2050,22 @@ public class StreamTaskTest {
final ProcessorTopology topology = withSources(asList(), mkMap());
final TopologyException exception = assertThrows(
TopologyException.class,
() -> new StreamTask(
taskId,
partitions,
topology,
consumer,
createConfig(false, "100"),
metrics,
stateDirectory,
cache,
time,
stateManager,
recordCollector,
context
)
);
TopologyException.class,
() -> new StreamTask(
taskId,
partitions,
topology,
consumer,
createConfig(false, "100"),
metrics,
stateDirectory,
cache,
time,
stateManager,
recordCollector,
context
)
);
assertThat(exception.getMessage(), equalTo("Invalid topology: " +
"Topic is unkown to the topology. This may happen if different KafkaStreams instances of the same " +