Github user siyuanh commented on a diff in the pull request:
https://github.com/apache/apex-malhar/pull/321#discussion_r68420530
--- Diff:
kafka/src/test/java/org/apache/apex/malhar/kafka/KafkaInputOperatorTest.java ---
@@ -264,49 +336,57 @@ public void testInputOperator(boolean hasFailure,
boolean idempotent) throws Exc
collector.isIdempotentTest = idempotent;
// Connect ports
- dag.addStream("Kafka message", node.outputPort,
collector.inputPort).setLocality(Locality.CONTAINER_LOCAL);
-
+ dag.addStream("Kafka message" + testName, node.outputPort,
collector.inputPort).setLocality(Locality.CONTAINER_LOCAL);
if (hasFailure) {
setupHasFailureTest(node, dag);
}
// Create local cluster
- final LocalMode.Controller lc = lma.getController();
+ LocalMode.Controller lc = lma.getController();
lc.setHeartbeatMonitoringEnabled(false);
- lc.runAsync();
-
- // Wait 30s for consumer finish consuming all the messages
- boolean notTimeout = latch.await(40000, TimeUnit.MILLISECONDS);
- Collections.sort(tupleCollection, new Comparator<String>()
- {
- @Override
- public int compare(String o1, String o2)
- {
- return Integer.parseInt(o1.split("_")[1]) -
Integer.parseInt(o2.split("_")[1]);
- }
- });
- Assert.assertTrue("TIMEOUT: 40s Collected " + tupleCollection,
notTimeout);
+ //let the Controller to run the inside another thread. It is almost
same as call Controller.runAsync(),
+ //but Controller.runAsync() don't expose the thread which run it, so
we don't know when the thread will be terminated.
+ //create this thread and then call join() to make sure the Controller
shutdown completely.
+ monitorThread = new Thread((StramLocalCluster)lc, "master");
+ monitorThread.start();
+
+ boolean notTimeout = true;
+ try {
+ // Wait 60s for consumer finish consuming all the messages
+ notTimeout = latch.await(waitTime, TimeUnit.MILLISECONDS);
+ lc.shutdown();
+
+ //wait until control thread finished.
+ monitorThread.join();
+ } catch (Exception e) {
+ logger.warn(e.getMessage());
+ }
+
+ t.join();
+
+ logger.info("Number of received/expected tuples: {}/{}, testName: {},
tuples: \n{}", tupleCollection.size(), expectedReceiveCount, testName,
tupleCollection);
--- End diff --
Log this only when something goes wrong
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
