You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2014/05/06 17:53:48 UTC
[1/2] git commit: STORM-296: fixed platform dependent tests
Repository: incubator-storm
Updated Branches:
refs/heads/master c1eb423c9 -> 5b4bb3135
STORM-296: fixed platform dependent tests
* removed hard coded test broker port
* cleanup temp zookeeper files
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/45b28482
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/45b28482
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/45b28482
Branch: refs/heads/master
Commit: 45b28482628a37d826a7c6a834d83bcc156a4a87
Parents: dc4de42
Author: wurstmeister <wu...@users.noreply.github.com>
Authored: Sat Apr 26 18:56:19 2014 +0100
Committer: wurstmeister <wu...@users.noreply.github.com>
Committed: Sat Apr 26 20:08:49 2014 +0100
----------------------------------------------------------------------
.../src/test/storm/kafka/KafkaTestBroker.java | 41 +++++++++++++-------
.../src/test/storm/kafka/KafkaUtilsTest.java | 8 +++-
.../src/test/storm/kafka/ZkCoordinatorTest.java | 2 +-
3 files changed, 35 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/45b28482/external/storm-kafka/src/test/storm/kafka/KafkaTestBroker.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/KafkaTestBroker.java b/external/storm-kafka/src/test/storm/kafka/KafkaTestBroker.java
index db21aaa..6966834 100644
--- a/external/storm-kafka/src/test/storm/kafka/KafkaTestBroker.java
+++ b/external/storm-kafka/src/test/storm/kafka/KafkaTestBroker.java
@@ -2,10 +2,16 @@ package storm.kafka;
import com.netflix.curator.framework.CuratorFramework;
import com.netflix.curator.framework.CuratorFrameworkFactory;
+import com.netflix.curator.framework.imps.CuratorFrameworkState;
import com.netflix.curator.retry.ExponentialBackoffRetry;
+import com.netflix.curator.test.InstanceSpec;
import com.netflix.curator.test.TestingServer;
+
+import kafka.server.KafkaConfig;
import kafka.server.KafkaServerStartable;
+import org.apache.commons.io.FileUtils;
+import java.io.File;
import java.io.IOException;
import java.util.Properties;
@@ -15,25 +21,22 @@ import java.util.Properties;
*/
public class KafkaTestBroker {
- private final int port = 49123;
+ private int port;
private KafkaServerStartable kafka;
private TestingServer server;
- private String zookeeperConnectionString;
+ private CuratorFramework zookeeper;
+ private File logDir;
public KafkaTestBroker() {
try {
server = new TestingServer();
- zookeeperConnectionString = server.getConnectString();
+ String zookeeperConnectionString = server.getConnectString();
ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
- String tempDir = System.getProperty("java.io.tmpdir");
- CuratorFramework zookeeper = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
+ zookeeper = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
zookeeper.start();
- Properties p = new Properties();
- p.setProperty("zookeeper.connect", zookeeperConnectionString);
- p.setProperty("broker.id", "0");
- p.setProperty("port", "" + port);
- p.setProperty("log.dirs",tempDir+"kafka_tmp_logs");
- kafka.server.KafkaConfig config = new kafka.server.KafkaConfig(p);
+ port = InstanceSpec.getRandomPort();
+ logDir = new File(System.getProperty("java.io.tmpdir"), "kafka/logs/kafka-test-" + port);
+ KafkaConfig config = buildKafkaConfig(zookeeperConnectionString);
kafka = new KafkaServerStartable(config);
kafka.startup();
} catch (Exception ex) {
@@ -41,6 +44,15 @@ public class KafkaTestBroker {
}
}
+ private kafka.server.KafkaConfig buildKafkaConfig(String zookeeperConnectionString) {
+ Properties p = new Properties();
+ p.setProperty("zookeeper.connect", zookeeperConnectionString);
+ p.setProperty("broker.id", "0");
+ p.setProperty("port", "" + port);
+ p.setProperty("log.dirs", logDir.getAbsolutePath());
+ return new KafkaConfig(p);
+ }
+
public String getBrokerConnectionString() {
return "localhost:" + port;
}
@@ -48,13 +60,16 @@ public class KafkaTestBroker {
public int getPort() {
return port;
}
-
public void shutdown() {
kafka.shutdown();
+ if (zookeeper.getState().equals(CuratorFrameworkState.STARTED)) {
+ zookeeper.close();
+ }
try {
- server.stop();
+ server.close();
} catch (IOException e) {
e.printStackTrace();
}
+ FileUtils.deleteQuietly(logDir);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/45b28482/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
index 25fa618..d68e802 100644
--- a/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
+++ b/external/storm-kafka/src/test/storm/kafka/KafkaUtilsTest.java
@@ -57,7 +57,11 @@ public class KafkaUtilsTest {
int port = broker.getPort();
broker.shutdown();
SimpleConsumer simpleConsumer = new SimpleConsumer("localhost", port, 100, 1024, "testClient");
- KafkaUtils.fetchMessages(config, simpleConsumer, new Partition(Broker.fromString(broker.getBrokerConnectionString()), 0), OffsetRequest.LatestTime());
+ try {
+ KafkaUtils.fetchMessages(config, simpleConsumer, new Partition(Broker.fromString(broker.getBrokerConnectionString()), 0), OffsetRequest.LatestTime());
+ } finally {
+ simpleConsumer.close();
+ }
}
@Test
@@ -173,7 +177,7 @@ public class KafkaUtilsTest {
private void createTopicAndSendMessage(String key, String value) {
Properties p = new Properties();
- p.setProperty("metadata.broker.list", "localhost:49123");
+ p.setProperty("metadata.broker.list", broker.getBrokerConnectionString());
p.setProperty("serializer.class", "kafka.serializer.StringEncoder");
ProducerConfig producerConfig = new ProducerConfig(p);
Producer<String, String> producer = new Producer<String, String>(producerConfig);
http://git-wip-us.apache.org/repos/asf/incubator-storm/blob/45b28482/external/storm-kafka/src/test/storm/kafka/ZkCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/external/storm-kafka/src/test/storm/kafka/ZkCoordinatorTest.java b/external/storm-kafka/src/test/storm/kafka/ZkCoordinatorTest.java
index c08ce82..2dd4d21 100644
--- a/external/storm-kafka/src/test/storm/kafka/ZkCoordinatorTest.java
+++ b/external/storm-kafka/src/test/storm/kafka/ZkCoordinatorTest.java
@@ -61,7 +61,7 @@ public class ZkCoordinatorTest {
public void shutdown() throws Exception {
simpleConsumer.close();
broker.shutdown();
- server.stop();
+ server.close();
}
@Test
[2/2] git commit: Merge branch 'STORM-296' of
https://github.com/wurstmeister/incubator-storm
Posted by bo...@apache.org.
Merge branch 'STORM-296' of https://github.com/wurstmeister/incubator-storm
STORM-296: Storm kafka unit tests are failing on windows (try 2)
Project: http://git-wip-us.apache.org/repos/asf/incubator-storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-storm/commit/5b4bb313
Tree: http://git-wip-us.apache.org/repos/asf/incubator-storm/tree/5b4bb313
Diff: http://git-wip-us.apache.org/repos/asf/incubator-storm/diff/5b4bb313
Branch: refs/heads/master
Commit: 5b4bb313510b1986afbe3e9522011f6f8376ca2c
Parents: c1eb423 45b2848
Author: Robert (Bobby) Evans <bo...@apache.org>
Authored: Tue May 6 15:52:37 2014 +0000
Committer: Robert (Bobby) Evans <bo...@apache.org>
Committed: Tue May 6 15:52:37 2014 +0000
----------------------------------------------------------------------
.../src/test/storm/kafka/KafkaTestBroker.java | 41 +++++++++++++-------
.../src/test/storm/kafka/KafkaUtilsTest.java | 8 +++-
.../src/test/storm/kafka/ZkCoordinatorTest.java | 2 +-
3 files changed, 35 insertions(+), 16 deletions(-)
----------------------------------------------------------------------