Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2108#discussion_r69130106
--- Diff:
flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/AtLeastOnceProducerTest.java
---
@@ -107,15 +115,16 @@ public void run() {
threadB.start();
// this should block:
producer.snapshotState(0, 0);
- // once all pending callbacks are confirmed, we can set this
marker to true
- markOne.set(true);
- for(int i = 0; i < 99; i++) {
- producer.invoke("msg-" + i);
+ synchronized (threadA) {
+ threadA.notifyAll(); // just in case, to let the test
fail faster
}
- // wait at most one second
- threadB.join(800L);
- Assert.assertFalse("Thread A reached this point too fast",
threadB.isAlive());
- if(runnableError.f0 != null) {
+
+ Deadline deadline = FiniteDuration.apply(5, "s").fromNow();
+ while (deadline.hasTimeLeft() && threadB.isAlive()) {
+ threadB.join(500);
+ }
+ Assert.assertFalse("Thread A is expected to be finished at this
point. If not, the test is prone to fail", threadB.isAlive());
+ if (runnableError.f0 != null) {
runnableError.f0.printStackTrace();
Assert.fail("Error from thread B: " + runnableError.f0
);
--- End diff --
Printing the stack trace to stdout is imo not so good. The problem is that
the stack trace will be intermingled with the rest of the testing log output. I
think it's better to simply rethrow the `Throwable` here.
---
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.
---
