[hotfix][kafka,test] Make brokers list final and avoid potential null…

… pointer exceptions
Pance · May 7, 2019 · 3b1976a · 3b1976a
1 parent 5b49f17
commit 3b1976a
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 6 deletions.
diff --git a/...1/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/...1/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -73,10 +73,10 @@
 public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
 
  protected static final Logger LOG = LoggerFactory.getLogger(KafkaTestEnvironmentImpl.class);
+ private final List<KafkaServer> brokers = new ArrayList<>();
  private File tmpZkDir;
  private File tmpKafkaParent;
  private List<File> tmpKafkaDirs;
- private List<KafkaServer> brokers;
  private TestingServer zookeeper;
  private String zookeeperConnectionString;
  private String brokerConnectionString = "";
@@ -258,14 +258,13 @@ public void prepare(Config config) throws Exception {
  }
 
  zookeeper = null;
- brokers = null;
+ brokers.clear();
 
  zookeeper = new TestingServer(-1, tmpZkDir);
  zookeeperConnectionString = zookeeper.getConnectString();
  LOG.info("Starting Zookeeper with zookeeperConnectionString: {}", zookeeperConnectionString);
 
  LOG.info("Starting KafkaServer");
- brokers = new ArrayList<>(config.getKafkaServersNumber());
 
  ListenerName listenerName = ListenerName.forSecurityProtocol(config.isSecureMode() ? SecurityProtocol.SASL_PLAINTEXT : SecurityProtocol.PLAINTEXT);
  for (int i = 0; i < config.getKafkaServersNumber(); i++) {

diff --git a/...a/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/...a/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java
@@ -74,10 +74,10 @@
 public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment {
 
  protected static final Logger LOG = LoggerFactory.getLogger(KafkaTestEnvironmentImpl.class);
+ private final List<KafkaServer> brokers = new ArrayList<>();
  private File tmpZkDir;
  private File tmpKafkaParent;
  private List<File> tmpKafkaDirs;
- private List<KafkaServer> brokers;
  private TestingServer zookeeper;
  private String zookeeperConnectionString;
  private String brokerConnectionString = "";
@@ -117,14 +117,13 @@ public void prepare(Config config) throws Exception {
  }
 
  zookeeper = null;
- brokers = null;
+ brokers.clear();
 
  zookeeper = new TestingServer(-1, tmpZkDir);
  zookeeperConnectionString = zookeeper.getConnectString();
  LOG.info("Starting Zookeeper with zookeeperConnectionString: {}", zookeeperConnectionString);
 
  LOG.info("Starting KafkaServer");
- brokers = new ArrayList<>(config.getKafkaServersNumber());
 
  ListenerName listenerName = ListenerName.forSecurityProtocol(config.isSecureMode() ? SecurityProtocol.SASL_PLAINTEXT : SecurityProtocol.PLAINTEXT);
  for (int i = 0; i < config.getKafkaServersNumber(); i++) {