KAFKA-4648: Improve test coverage StreamTask #2451
Conversation
dguy
changed the title
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
LGTM |
| @Test(expected = IllegalStateException.class) | ||
| public void shouldThrowIllegalStateExceptionIfCurrentNodeIsNotNullWhenPunctuateCalled() throws Exception { | ||
| ((ProcessorContextImpl) task.processorContext()).setCurrentNode(processor); | ||
| task.punctuate(processor, 10); |
There was a problem hiding this comment.
wrap with try-catch instead of expected annotation -- more than one line test.
| @SuppressWarnings("unchecked") | ||
| @Test | ||
| public void shouldCloseAllProcessorNodesWhenExceptionsRaised() throws Exception { | ||
| final MockSourceNode processorNode = new MockSourceNode(topic1, intDeserializer, intDeserializer) { |
There was a problem hiding this comment.
Unify "create task" code with shouldThrowExceptionIfAnyExceptionsRaisedDuringCloseTopology -- it's almost the same and both test cases can use the same topology structure.
|
resolved merge conflicts and addressed feedback. @guozhangwang can you please merge? |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
| final StreamTask streamTask = new StreamTask(new TaskId(0, 0), "applicationId", partitions, | ||
| topology, consumer, restoreStateConsumer, config, streamsMetrics, stateDirectory, new ThreadCache("testCache", 0, streamsMetrics), new MockTime(), recordCollector); | ||
|
|
||
| task = new StreamTask(taskId00, applicationId, partitions, |
There was a problem hiding this comment.
Could we just initialize the task in setUp with the testCache so that we do not need to explicit re-initialize it here? If not we need to at least close the task before re-initialize it.
There was a problem hiding this comment.
It needs the topology that is constructed just above, so no.
| @SuppressWarnings("unchecked") | ||
| @Test | ||
| public void shouldThrowExceptionIfAnyExceptionsRaisedDuringCloseTopology() throws Exception { | ||
| task = createTaskThatThrowsExceptionOnClose(); |
There was a problem hiding this comment.
Should we close the task first before re-initialize it to another StreamTask? Ditto below.
There was a problem hiding this comment.
it doesn't really matter. It is not like the topology has created any state stores, but if it makes you feel better...
| final List<ProcessorNode> processorNodes = Arrays.asList(processorNode, processor, source1, source2); | ||
| final Map<String, SourceNode> sourceNodes | ||
| = Collections.<String, SourceNode>singletonMap(topic1[0], processorNode); | ||
| final ProcessorTopology topology = new ProcessorTopology(processorNodes, |
There was a problem hiding this comment.
Could we reuse the class's own private topology here?
There was a problem hiding this comment.
No - the topology is different.
|
updated |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
Provide test coverage for exception paths in:
schedule(),closeTopology(), andpunctuate()