[FLINK-14680] Enable KafkaConsumerTestBase#runFailOnNoBrokerTest to pass with new DefaultScheduler

• Type: Sub-task
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.10.0
• Fix Version/s: 1.10.0
• Component/s: Runtime / Coordination, Tests
• Labels: pull-request-available
• Assignee: Gary Yao
• Reporter: Gary Yao
• Created: 08/Nov/19 13:34
• Updated: 14/Nov/19 16:41
• Resolved: 14/Nov/19 16:41

Description

KafkaConsumerTestBase#runFailOnNoBrokerTest has assumptions on the causal chain of the JobExecutionException. In particular, it assumes that the exception caused by user code is the direct cause of JobExecutionException. However, this is no longer true when using the DefaultScheduler, which wraps the exception in an JobException, which additionally specifies the reason of the job recovery suppression.

The code in question is listed below:

		} catch (JobExecutionException jee) {
			if (kafkaServer.getVersion().equals("0.9") ||
				kafkaServer.getVersion().equals("0.10") ||
				kafkaServer.getVersion().equals("0.11") ||
				kafkaServer.getVersion().equals("2.0")) {
				assertTrue(jee.getCause() instanceof TimeoutException);

				TimeoutException te = (TimeoutException) jee.getCause();

				assertEquals("Timeout expired while fetching topic metadata", te.getMessage());
			} else {
				assertTrue(jee.getCause() instanceof RuntimeException);

				RuntimeException re = (RuntimeException) jee.getCause();

				assertTrue(re.getMessage().contains("Unable to retrieve any partitions"));
			}
		}

Issue Links

• links to: GitHub Pull Request #10136