You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/12 14:50:34 UTC
[3/5] flink git commit: [hotfix][kafka-tests] Do not hide original
exception in FlinkKafkaProducer011ITCase
[hotfix][kafka-tests] Do not hide original exception in FlinkKafkaProducer011ITCase
This closes #5383.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/02ea5087
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/02ea5087
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/02ea5087
Branch: refs/heads/master
Commit: 02ea508736e06e08ff53921994b29ce8db0f5f50
Parents: 1c02e1a
Author: Piotr Nowojski <pi...@gmail.com>
Authored: Tue Jan 30 14:56:43 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Mon Feb 12 11:59:30 2018 +0100
----------------------------------------------------------------------
.../kafka/FlinkKafkaProducer011ITCase.java | 35 ++++++++------------
1 file changed, 13 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/02ea5087/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
----------------------------------------------------------------------
diff --git a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
index 3c3c86a..ab3a2e2 100644
--- a/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
+++ b/flink-connectors/flink-connector-kafka-0.11/src/test/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer011ITCase.java
@@ -39,15 +39,15 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Optional;
import java.util.Properties;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static org.apache.flink.util.ExceptionUtils.findThrowable;
import static org.apache.flink.util.Preconditions.checkState;
-import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.fail;
/**
* IT cases for the {@link FlinkKafkaProducer011}.
@@ -98,7 +98,9 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
throw new IllegalStateException("This should not be reached.");
}
catch (Exception ex) {
- assertIsCausedBy(FlinkKafka011ErrorCode.PRODUCERS_POOL_EMPTY, ex);
+ if (!isCausedBy(FlinkKafka011ErrorCode.PRODUCERS_POOL_EMPTY, ex)) {
+ throw ex;
+ }
}
// Resume transactions before testHarness1 is being closed (in case of failures close() might not be called)
@@ -114,7 +116,9 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
}
catch (Exception ex) {
// testHarness1 will be fenced off after creating and closing testHarness2
- assertIsCausedBy(ProducerFencedException.class, ex);
+ if (!findThrowable(ex, ProducerFencedException.class).isPresent()) {
+ throw ex;
+ }
}
}
@@ -592,24 +596,11 @@ public class FlinkKafkaProducer011ITCase extends KafkaTestBase {
return properties;
}
- private void assertIsCausedBy(Class<?> clazz, Throwable ex) {
- for (int depth = 0; depth < 50 && ex != null; depth++) {
- if (clazz.isInstance(ex)) {
- return;
- }
- ex = ex.getCause();
- }
- fail(String.format("Exception [%s] was not caused by [%s]", ex, clazz));
- }
-
- private void assertIsCausedBy(FlinkKafka011ErrorCode expectedErrorCode, Throwable ex) {
- for (int depth = 0; depth < 50 && ex != null; depth++) {
- if (ex instanceof FlinkKafka011Exception) {
- assertEquals(expectedErrorCode, ((FlinkKafka011Exception) ex).getErrorCode());
- return;
- }
- ex = ex.getCause();
+ private boolean isCausedBy(FlinkKafka011ErrorCode expectedErrorCode, Throwable ex) {
+ Optional<FlinkKafka011Exception> cause = findThrowable(ex, FlinkKafka011Exception.class);
+ if (cause.isPresent()) {
+ return cause.get().getErrorCode().equals(expectedErrorCode);
}
- fail(String.format("Exception [%s] was not caused by FlinkKafka011Exception[errorCode=%s]", ex, expectedErrorCode));
+ return false;
}
}