zkServers = null;
+
+ /** ZooKeeper port */
public Integer zkPort = null;
+
+ /** ZooKeeper context root */
public String zkRoot = null;
+
+ /** Spout id which is also used to track the Kafka offsets */
public String id = null;
public String outputStreamId;
- // setting for how often to save the current kafka offset to ZooKeeper
+ /** setting for how often to save the current kafka offset to ZooKeeper */
public long stateUpdateIntervalMs = 2000;
// Exponential back-off retry settings. These are used when retrying messages after a bolt
@@ -38,6 +49,20 @@ public class SpoutConfig extends KafkaConfig implements Serializable {
public double retryDelayMultiplier = 1.0;
public long retryDelayMaxMs = 60 * 1000;
+ /** offset state information storage. validate options are zookeeper and kafka */
+ public String stateStore = STATE_STORE_ZOOKEEPER;
+
+ /** timeout in millis for state read/write operations */
+ public int stateOpTimeout = 5000;
+
+ /** max retries allowed for state read/write operations */
+ public int stateOpMaxRetry = 3;
+
+ public SpoutConfig(BrokerHosts hosts, String topic, String id) {
+ super(hosts, topic);
+ this.id = id;
+ }
+
public SpoutConfig(BrokerHosts hosts, String topic, String zkRoot, String id) {
super(hosts, topic);
this.zkRoot = zkRoot;
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/StateStore.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/StateStore.java
new file mode 100644
index 00000000000..60103bf7d29
--- /dev/null
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/StateStore.java
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.kafka;
+
+import java.io.Closeable;
+import java.util.Map;
+
+/**
+ * Abstraction of a partition state storage.
+ *
+ * The partition state usually is kept in Json format in the store and in Map format in runtime memory. An example
+ * is shown below:
+ *
+ *
+ *
+ * Json:
+ *
+ * {
+ * "broker": {
+ * "host": "kafka.sample.net",
+ * "port": 9092
+ * },
+ * "offset": 4285,
+ * "partition": 1,
+ * "topic": "testTopic",
+ * "topology": {
+ * "id": "fce905ff-25e0 -409e-bc3a-d855f 787d13b",
+ * "name": "Test Topology"
+ * }
+ * }
+ *
+ *
+ *
+ *
+ * Memory:
+ *
+ * Map data = (Map) ImmutableMap.builder()
+ * .put("topology", ImmutableMap.of(
+ * "id", "fce905ff-25e0 -409e-bc3a-d855f 787d13b",
+ * "name", "Test Topology"))
+ * .put("offset", 4285)
+ * .put("partition", 1)
+ * .put("broker", ImmutableMap.of(
+ * "host", "kafka.sample.net",
+ * "port", 9092))
+ * .put("topic", "testTopic").build();
+ *
+ *
+ *
+ *
+ *
+ * User can create their own custom state store by implementing this interface and register it with
+ * {@link SpoutConfig#stateStore}. The implementation class must also provide a public constructor
+ * that takes two arguments:
+ *
+ *
+ *
+ * public CustomStateStor(Map stormConf and SpoutConfig spoutConfig)
+ *
+ *
+ *
+ * See {@see KafkaStateStore} class as an example.
+ *
+ */
+public interface StateStore extends Closeable {
+
+ Map readState(Partition p);
+
+ void writeState(Partition p, Map state);
+}
\ No newline at end of file
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/StaticCoordinator.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/StaticCoordinator.java
index bdbc44d2b15..00602490b5f 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/StaticCoordinator.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/StaticCoordinator.java
@@ -23,16 +23,18 @@
public class StaticCoordinator implements PartitionCoordinator {
- Map _managers = new HashMap();
- List _allManagers = new ArrayList();
- public StaticCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig config, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId) {
- StaticHosts hosts = (StaticHosts) config.hosts;
+ private Map _managers = new HashMap();
+ private List _allManagers = new ArrayList();
+
+ public StaticCoordinator(DynamicPartitionConnections connections, PartitionStateManagerFactory partitionStateManagerFactory, Map stormConf, SpoutConfig spoutConfig, int taskIndex, int totalTasks, String topologyInstanceId) {
+ StaticHosts hosts = (StaticHosts) spoutConfig.hosts;
List partitions = new ArrayList();
partitions.add(hosts.getPartitionInformation());
List myPartitions = KafkaUtils.calculatePartitionsForTask(partitions, totalTasks, taskIndex);
+
for (Partition myPartition : myPartitions) {
- _managers.put(myPartition, new PartitionManager(connections, topologyInstanceId, state, stormConf, config, myPartition));
+ _managers.put(myPartition, new PartitionManager(connections, topologyInstanceId, partitionStateManagerFactory.getInstance(myPartition), stormConf, spoutConfig, myPartition));
}
_allManagers = new ArrayList(_managers.values());
}
@@ -48,5 +50,4 @@ public PartitionManager getManager(Partition partition) {
@Override
public void refresh() { return; }
-
}
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java
index a53d5660a04..fa35626a971 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkCoordinator.java
@@ -26,7 +26,7 @@
import static org.apache.storm.kafka.KafkaUtils.taskId;
public class ZkCoordinator implements PartitionCoordinator {
- public static final Logger LOG = LoggerFactory.getLogger(ZkCoordinator.class);
+ private static final Logger LOG = LoggerFactory.getLogger(ZkCoordinator.class);
SpoutConfig _spoutConfig;
int _taskIndex;
@@ -34,25 +34,25 @@ public class ZkCoordinator implements PartitionCoordinator {
String _topologyInstanceId;
Map _managers = new HashMap();
List _cachedList = new ArrayList();
+ PartitionStateManagerFactory _partitionStateManagerFactory;
Long _lastRefreshTime = null;
int _refreshFreqMs;
DynamicPartitionConnections _connections;
DynamicBrokersReader _reader;
- ZkState _state;
Map _stormConf;
- public ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId) {
- this(connections, stormConf, spoutConfig, state, taskIndex, totalTasks, topologyInstanceId, buildReader(stormConf, spoutConfig));
+ public ZkCoordinator(DynamicPartitionConnections connections, PartitionStateManagerFactory partitionStateManagerFactory, Map stormConf, SpoutConfig spoutConfig, int taskIndex, int totalTasks, String topologyInstanceId) {
+ this(connections, partitionStateManagerFactory, stormConf, spoutConfig, taskIndex, totalTasks, topologyInstanceId, buildReader(stormConf, spoutConfig));
}
- public ZkCoordinator(DynamicPartitionConnections connections, Map stormConf, SpoutConfig spoutConfig, ZkState state, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader) {
- _spoutConfig = spoutConfig;
+ public ZkCoordinator(DynamicPartitionConnections connections, PartitionStateManagerFactory partitionStateManagerFactory, Map stormConf, SpoutConfig spoutConfig, int taskIndex, int totalTasks, String topologyInstanceId, DynamicBrokersReader reader) {
_connections = connections;
+ _partitionStateManagerFactory = partitionStateManagerFactory;
_taskIndex = taskIndex;
_totalTasks = totalTasks;
_topologyInstanceId = topologyInstanceId;
_stormConf = stormConf;
- _state = state;
+ _spoutConfig = spoutConfig;
ZkHosts brokerConf = (ZkHosts) spoutConfig.hosts;
_refreshFreqMs = brokerConf.refreshFreqSecs * 1000;
_reader = reader;
@@ -95,7 +95,7 @@ public void refresh() {
LOG.info(taskId(_taskIndex, _totalTasks) + "New partition managers: " + newPartitions.toString());
for (Partition id : newPartitions) {
- PartitionManager man = new PartitionManager(_connections, _topologyInstanceId, _state, _stormConf, _spoutConfig, id);
+ PartitionManager man = new PartitionManager(_connections, _topologyInstanceId, _partitionStateManagerFactory.getInstance(id), _stormConf, _spoutConfig, id);
_managers.put(id, man);
}
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkStateStore.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkStateStore.java
new file mode 100644
index 00000000000..ba4c5857a8f
--- /dev/null
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/ZkStateStore.java
@@ -0,0 +1,192 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.kafka;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.storm.Config;
+import org.apache.storm.utils.Utils;
+import org.apache.zookeeper.CreateMode;
+import org.json.simple.JSONValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.Map;
+
+public class ZkStateStore implements StateStore {
+ private static final Logger LOG = LoggerFactory.getLogger(ZkStateStore.class);
+
+ private ZkStateStoreConfig _config;
+ private CuratorFramework _curator;
+
+ private CuratorFramework newCurator(ZkStateStoreConfig config) throws Exception {
+ LOG.info("Creating new curator framework on {}.", config.getZkServerPorts());
+ return CuratorFrameworkFactory.newClient(config.getZkServerPorts(),
+ config.getSessionTimeout(), config.getConnectionTimeout(),
+ new RetryNTimes(config.getRetryTimes(), config.getRetryInterval()));
+ }
+
+ public CuratorFramework getCurator() {
+ assert _curator != null;
+ return _curator;
+ }
+
+ public ZkStateStore(ZkStateStoreConfig config) {
+ this._config = config;
+ try {
+ _curator = newCurator(_config);
+ _curator.start();
+ LOG.info("Started curator framework.");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public ZkStateStore(Map conf, SpoutConfig spoutConfig) {
+ this(new ZkStateStoreConfig(conf, spoutConfig));
+ }
+
+ @Override
+ public void writeState(Partition p, Map state) {
+ String zkPath = committedPath(p);
+ LOG.debug("Writing to {} with stat data {} for partition {}:{}.", zkPath, state, p.host, p.partition);
+ write(zkPath, JSONValue.toJSONString(state).getBytes(Charset.forName("UTF-8")));
+ }
+
+ @Override
+ public Map readState(Partition p) {
+ String zkPath = committedPath(p);
+ LOG.debug("Reading from {} for state data for partition {}:{}.", zkPath, p.host, p.partition);
+ try {
+ byte[] b = read(zkPath);
+ if (b == null) {
+ LOG.warn("No state found for partition {}:{} at this time.", p.host, p.partition);
+ return null;
+ }
+ Map state = (Map) JSONValue.parse(new String(b, "UTF-8"));
+ LOG.debug("Retrieved state {} from {} for partition {}:{}.", state, zkPath, p.host, p.partition);
+ return state;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void close() {
+ _curator.close();
+ _curator = null;
+ LOG.info("Closed curator framework.");
+ }
+
+ private String committedPath(Partition partition) {
+ return _config.getZkRoot() + "/" + _config.getConsumerId() + "/" + partition.getId();
+ }
+
+ private void write(String path, byte[] bytes) {
+ try {
+ if (_curator.checkExists().forPath(path) == null) {
+ _curator.create()
+ .creatingParentsIfNeeded()
+ .withMode(CreateMode.PERSISTENT)
+ .forPath(path, bytes);
+ } else {
+ _curator.setData().forPath(path, bytes);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private byte[] read(String path) {
+ try {
+ if (_curator.checkExists().forPath(path) != null) {
+ return _curator.getData().forPath(path);
+ } else {
+ return null;
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static class ZkStateStoreConfig {
+ private final String zkServerPorts;
+ private final String zkRoot;
+ private final String consumerId;
+ private final int connectionTimeout;
+ private final int sessionTimeout;
+ private final int retryTimes;
+ private final int retryInterval;
+
+ public ZkStateStoreConfig(Map conf, SpoutConfig spoutConfig) {
+ List zkServers = spoutConfig.zkServers;
+ if (zkServers == null) {
+ zkServers = (List) conf.get(Config.STORM_ZOOKEEPER_SERVERS);
+ }
+
+ Integer zkPort = spoutConfig.zkPort;
+ if (zkPort == null) {
+ zkPort = ((Number) conf.get(Config.STORM_ZOOKEEPER_PORT)).intValue();
+ }
+
+ String serverPorts = "";
+ for (String server : zkServers) {
+ serverPorts = serverPorts + server + ":" + zkPort + ",";
+ }
+
+ this.zkServerPorts = serverPorts;
+ this.zkRoot = spoutConfig.zkRoot;
+ this.consumerId = spoutConfig.id;
+ this.connectionTimeout = Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT));
+ this.sessionTimeout = Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT));
+ this.retryTimes = Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_TIMES));
+ this.retryInterval = Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL));
+ }
+
+ public String getZkServerPorts() {
+ return zkServerPorts;
+ }
+
+ public String getZkRoot() {
+ return zkRoot;
+ }
+
+ public String getConsumerId() {
+ return consumerId;
+ }
+
+ public int getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ public int getSessionTimeout() {
+ return sessionTimeout;
+ }
+
+ public int getRetryTimes() {
+ return retryTimes;
+ }
+
+ public int getRetryInterval() {
+ return retryInterval;
+ }
+ }
+}
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
index 9732c8c57b4..1ecb4152d81 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/TridentKafkaEmitter.java
@@ -41,7 +41,7 @@
public class TridentKafkaEmitter {
- public static final Logger LOG = LoggerFactory.getLogger(TridentKafkaEmitter.class);
+ private static final Logger LOG = LoggerFactory.getLogger(TridentKafkaEmitter.class);
private DynamicPartitionConnections _connections;
private String _topologyName;
@@ -100,7 +100,7 @@ private Map doEmitNewPartitionBatch(SimpleConsumer consumer, Partition partition
if (lastTopoMeta != null) {
lastInstanceId = (String) lastTopoMeta.get("id");
}
- if (_config.ignoreZkOffsets && !_topologyInstanceId.equals(lastInstanceId)) {
+ if ((_config.ignoreStoredOffsets || _config.ignoreZkOffsets) && !_topologyInstanceId.equals(lastInstanceId)) {
offset = KafkaUtils.getOffset(consumer, partition.topic, partition.partition, _config.startOffsetTime);
} else {
offset = (Long) lastMeta.get("nextOffset");
@@ -157,7 +157,7 @@ private ByteBufferMessageSet fetchMessages(SimpleConsumer consumer, Partition pa
private void reEmitPartitionBatch(TransactionAttempt attempt, TridentCollector collector, Partition partition, Map meta) {
LOG.info("re-emitting batch, attempt " + attempt);
String instanceId = (String) meta.get("instanceId");
- if (!_config.ignoreZkOffsets || instanceId.equals(_topologyInstanceId)) {
+ if (!(_config.ignoreStoredOffsets || _config.ignoreZkOffsets) || instanceId.equals(_topologyInstanceId)) {
SimpleConsumer consumer = _connections.register(partition);
long offset = (Long) meta.get("offset");
long nextOffset = (Long) meta.get("nextOffset");
diff --git a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/ZkBrokerReader.java b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/ZkBrokerReader.java
index d26c341da8a..abc10db3d7d 100644
--- a/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/ZkBrokerReader.java
+++ b/external/storm-kafka/src/jvm/org/apache/storm/kafka/trident/ZkBrokerReader.java
@@ -29,7 +29,7 @@
public class ZkBrokerReader implements IBrokerReader {
- public static final Logger LOG = LoggerFactory.getLogger(ZkBrokerReader.class);
+ private static final Logger LOG = LoggerFactory.getLogger(ZkBrokerReader.class);
List cachedBrokers = new ArrayList();
DynamicBrokersReader reader;
diff --git a/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaStateStoreTest.java b/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaStateStoreTest.java
new file mode 100644
index 00000000000..6af2a1a3643
--- /dev/null
+++ b/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaStateStoreTest.java
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.kafka;
+
+import com.google.common.collect.ImmutableMap;
+import kafka.javaapi.producer.Producer;
+import kafka.producer.KeyedMessage;
+import kafka.producer.ProducerConfig;
+import org.apache.curator.test.TestingServer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+
+public class KafkaStateStoreTest {
+
+ private KafkaTestBroker testBroker;
+ private KafkaStateStore stateStore;
+
+ @Before
+ public void setUp() throws Exception {
+ String testTopic = "testTopic";
+
+ TestingServer server = new TestingServer();
+ testBroker = new KafkaTestBroker(server, 0);
+ String connectionString = server.getConnectString();
+
+ Properties props = new Properties();
+ props.put("metadata.broker.list", testBroker.getBrokerConnectionString());
+ Producer p = new Producer<>(new ProducerConfig(props));
+ KeyedMessage msg = new KeyedMessage<>(testTopic, "test message".getBytes());
+ p.send(msg);
+
+ ZkHosts hosts = new ZkHosts(connectionString);
+ SpoutConfig spoutConfig = new SpoutConfig(hosts, testTopic, "/", "testConsumerGroup");
+
+ Map stormConf = new HashMap();
+
+ stateStore = new KafkaStateStore(stormConf, spoutConfig);
+ }
+
+ @After
+ public void shutdown() throws Exception {
+ testBroker.shutdown();
+ }
+
+ @Test
+ public void testStoreReadWrite() {
+ Partition testPartition = new Partition(new Broker("localhost", testBroker.getPort()), "testTopic", 0);
+
+ Map broker = ImmutableMap.of("host", "kafka.sample.net", "port", 9100L);
+ Map topology = ImmutableMap.of("id", "fce905ff-25e0 -409e-bc3a-d855f 787d13b", "name", "Test Topology");
+ Map testState = ImmutableMap.of("broker", broker, "offset", 4285L, "partition", 1L, "topic", "testTopic", "topology", topology);
+
+ stateStore.writeState(testPartition, testState);
+ Map state = stateStore.readState(testPartition);
+
+ assertEquals("kafka.sample.net", ((Map)state.get("broker")).get("host"));
+ assertEquals(9100L, ((Map)state.get("broker")).get("port"));
+ assertEquals(4285L, state.get("offset"));
+ assertEquals(1L, state.get("partition"));
+ assertEquals(4285L, state.get("offset"));
+ assertEquals("testTopic", state.get("topic"));
+ assertEquals("fce905ff-25e0 -409e-bc3a-d855f 787d13b", ((Map) state.get("topology")).get("id"));
+ assertEquals("Test Topology", ((Map)state.get("topology")).get("name"));
+ }
+}
diff --git a/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaTestBroker.java b/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaTestBroker.java
index e2fb60f5a03..75d03480c2c 100644
--- a/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaTestBroker.java
+++ b/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaTestBroker.java
@@ -45,15 +45,19 @@ public class KafkaTestBroker {
private File logDir;
public KafkaTestBroker() {
+ this(null, 0);
+ }
+
+ public KafkaTestBroker(TestingServer s, int brokerId) {
try {
- server = new TestingServer();
+ server = s == null? server = new TestingServer() : s;
String zookeeperConnectionString = server.getConnectString();
ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
zookeeper = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
zookeeper.start();
port = InstanceSpec.getRandomPort();
logDir = new File(System.getProperty("java.io.tmpdir"), "kafka/logs/kafka-test-" + port);
- KafkaConfig config = buildKafkaConfig(zookeeperConnectionString);
+ KafkaConfig config = buildKafkaConfig(zookeeperConnectionString, brokerId);
kafka = new KafkaServerStartable(config);
kafka.startup();
} catch (Exception ex) {
@@ -61,12 +65,14 @@ public KafkaTestBroker() {
}
}
- private kafka.server.KafkaConfig buildKafkaConfig(String zookeeperConnectionString) {
+ private kafka.server.KafkaConfig buildKafkaConfig(String zookeeperConnectionString, int brokerId) {
Properties p = new Properties();
p.setProperty("zookeeper.connect", zookeeperConnectionString);
- p.setProperty("broker.id", "0");
+ p.setProperty("broker.id", String.valueOf(brokerId));
p.setProperty("port", "" + port);
p.setProperty("log.dirs", logDir.getAbsolutePath());
+ // we need to make it default to 1 so that offset manager can create the internal topic during unit test
+ p.setProperty("offsets.topic.replication.factor", "1");
return new KafkaConfig(p);
}
diff --git a/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaUtilsTest.java b/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaUtilsTest.java
index 9da6c0a5187..85756d999af 100644
--- a/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaUtilsTest.java
+++ b/external/storm-kafka/src/test/org/apache/storm/kafka/KafkaUtilsTest.java
@@ -117,7 +117,7 @@ public void fetchMessagesWithInvalidOffsetAndDefaultHandlingEnabled() throws Exc
@Test
public void getOffsetFromConfigAndDontForceFromStart() {
- config.ignoreZkOffsets = false;
+ config.ignoreStoredOffsets = false;
config.startOffsetTime = OffsetRequest.EarliestTime();
createTopicAndSendMessage();
long latestOffset = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, OffsetRequest.EarliestTime());
@@ -127,7 +127,7 @@ public void getOffsetFromConfigAndDontForceFromStart() {
@Test
public void getOffsetFromConfigAndFroceFromStart() {
- config.ignoreZkOffsets = true;
+ config.ignoreStoredOffsets = true;
config.startOffsetTime = OffsetRequest.EarliestTime();
createTopicAndSendMessage();
long earliestOffset = KafkaUtils.getOffset(simpleConsumer, config.topic, 0, OffsetRequest.EarliestTime());
diff --git a/external/storm-kafka/src/test/org/apache/storm/kafka/PartitionStateManagerFactoryTest.java b/external/storm-kafka/src/test/org/apache/storm/kafka/PartitionStateManagerFactoryTest.java
new file mode 100644
index 00000000000..4966ec3f672
--- /dev/null
+++ b/external/storm-kafka/src/test/org/apache/storm/kafka/PartitionStateManagerFactoryTest.java
@@ -0,0 +1,72 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.kafka;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+public class PartitionStateManagerFactoryTest {
+
+ PartitionStateManagerFactory factory;
+
+ @Before
+ public void setUp() throws Exception {
+ }
+
+ @Test
+ public void testCustomStateStore() throws Exception {
+ Map stormConfig = new HashMap();
+ SpoutConfig spoutConfig = new SpoutConfig(null, null, null);
+ spoutConfig.stateStore = TestStateStore.class.getName();
+
+ factory = new PartitionStateManagerFactory(stormConfig, spoutConfig);
+
+ PartitionStateManager partitionStateManager = factory.getInstance(null);
+
+ assertEquals(TestStateStore.MAGIC_STATE, partitionStateManager.getState());
+ }
+
+ public static class TestStateStore implements StateStore {
+
+ public static final Map MAGIC_STATE = ImmutableMap.of(new Object(), new Object());
+
+ public TestStateStore(Map stormConf, SpoutConfig spoutConfig) {}
+
+ @Override
+ public Map readState(Partition p) {
+ return MAGIC_STATE;
+ }
+
+ @Override
+ public void writeState(Partition p, Map state) {
+
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+ }
+}
\ No newline at end of file
diff --git a/external/storm-kafka/src/test/org/apache/storm/kafka/ZKStateStoreTest.java b/external/storm-kafka/src/test/org/apache/storm/kafka/ZKStateStoreTest.java
new file mode 100644
index 00000000000..39b1b74c338
--- /dev/null
+++ b/external/storm-kafka/src/test/org/apache/storm/kafka/ZKStateStoreTest.java
@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.storm.kafka;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.curator.test.TestingServer;
+import org.apache.storm.Config;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.MockitoAnnotations;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class ZKStateStoreTest {
+
+ private TestingServer server;
+ private ZkStateStore stateStore;
+
+ @Before
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+
+ server = new TestingServer();
+ String connectionString = server.getConnectString();
+ ZkHosts hosts = new ZkHosts(connectionString);
+
+ SpoutConfig spoutConfig;
+ spoutConfig = new SpoutConfig(hosts, "topic", "/test", "id");
+ spoutConfig.zkServers = Arrays.asList("localhost");
+ spoutConfig.zkPort = server.getPort();
+
+ Map stormConf = new HashMap<>();
+ stormConf.put(Config.STORM_ZOOKEEPER_PORT, spoutConfig.zkPort);
+ stormConf.put(Config.STORM_ZOOKEEPER_SERVERS, spoutConfig.zkServers);
+ stormConf.put(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT, 20000);
+ stormConf.put(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT, 20000);
+ stormConf.put(Config.STORM_ZOOKEEPER_RETRY_TIMES, 3);
+ stormConf.put(Config.STORM_ZOOKEEPER_RETRY_INTERVAL, 30);
+
+ stateStore = new ZkStateStore(stormConf, spoutConfig);
+ }
+
+ @After
+ public void shutdown() throws Exception {
+ stateStore.close();
+ server.close();
+ }
+
+ @Test
+ public void testStoreReadWrite() {
+
+ Partition testPartition = new Partition(new Broker("localhost", 9100), "testTopic", 1);
+
+ Map broker = ImmutableMap.of("host", "kafka.sample.net", "port", 9100L);
+ Map topology = ImmutableMap.of("id", "fce905ff-25e0 -409e-bc3a-d855f 787d13b", "name", "Test Topology");
+ Map testState = ImmutableMap.of("broker", broker, "offset", 4285L, "partition", 1L, "topic", "testTopic", "topology", topology);
+
+ stateStore.writeState(testPartition, testState);
+ Map state = stateStore.readState(testPartition);
+
+ assertEquals("kafka.sample.net", ((Map)state.get("broker")).get("host"));
+ assertEquals(9100L, ((Map)state.get("broker")).get("port"));
+ assertEquals(4285L, state.get("offset"));
+ assertEquals(1L, state.get("partition"));
+ assertEquals(4285L, state.get("offset"));
+ assertEquals("testTopic", state.get("topic"));
+ assertEquals("fce905ff-25e0 -409e-bc3a-d855f 787d13b", ((Map) state.get("topology")).get("id"));
+ assertEquals("Test Topology", ((Map)state.get("topology")).get("name"));
+ }
+}
diff --git a/external/storm-kafka/src/test/org/apache/storm/kafka/ZkCoordinatorTest.java b/external/storm-kafka/src/test/org/apache/storm/kafka/ZkCoordinatorTest.java
index 65bf0b45ee1..eff6c2e17e6 100644
--- a/external/storm-kafka/src/test/org/apache/storm/kafka/ZkCoordinatorTest.java
+++ b/external/storm-kafka/src/test/org/apache/storm/kafka/ZkCoordinatorTest.java
@@ -47,34 +47,35 @@ public class ZkCoordinatorTest {
private TestingServer server;
private Map stormConf = new HashMap();
private SpoutConfig spoutConfig;
- private ZkState state;
private SimpleConsumer simpleConsumer;
+ private PartitionStateManagerFactory partitionStateManagerFactory;
@Before
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
+
server = new TestingServer();
String connectionString = server.getConnectString();
ZkHosts hosts = new ZkHosts(connectionString);
hosts.refreshFreqSecs = 1;
+
spoutConfig = new SpoutConfig(hosts, "topic", "/test", "id");
- Map conf = buildZookeeperConfig(server);
- state = new ZkState(conf);
+ spoutConfig.zkServers = Arrays.asList("localhost");
+ spoutConfig.zkPort = server.getPort();
+
+ stormConf.put(Config.TRANSACTIONAL_ZOOKEEPER_PORT, server.getPort());
+ stormConf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Arrays.asList("localhost"));
+ stormConf.put(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT, 20000);
+ stormConf.put(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT, 20000);
+ stormConf.put(Config.STORM_ZOOKEEPER_RETRY_TIMES, 3);
+ stormConf.put(Config.STORM_ZOOKEEPER_RETRY_INTERVAL, 30);
+
+ partitionStateManagerFactory = new PartitionStateManagerFactory(stormConf, spoutConfig);
+
simpleConsumer = new SimpleConsumer("localhost", broker.getPort(), 60000, 1024, "testClient");
when(dynamicPartitionConnections.register(any(Broker.class), any(String.class) ,anyInt())).thenReturn(simpleConsumer);
}
- private Map buildZookeeperConfig(TestingServer server) {
- Map conf = new HashMap();
- conf.put(Config.TRANSACTIONAL_ZOOKEEPER_PORT, server.getPort());
- conf.put(Config.TRANSACTIONAL_ZOOKEEPER_SERVERS, Arrays.asList("localhost"));
- conf.put(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT, 20000);
- conf.put(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT, 20000);
- conf.put(Config.STORM_ZOOKEEPER_RETRY_TIMES, 3);
- conf.put(Config.STORM_ZOOKEEPER_RETRY_INTERVAL, 30);
- return conf;
- }
-
@After
public void shutdown() throws Exception {
simpleConsumer.close();
@@ -95,7 +96,6 @@ public void testOnePartitionPerTask() throws Exception {
}
}
-
@Test
public void testPartitionsChange() throws Exception {
final int totalTasks = 64;
@@ -138,7 +138,7 @@ private void waitForRefresh() throws InterruptedException {
private List buildCoordinators(int totalTasks) {
List coordinatorList = new ArrayList();
for (int i = 0; i < totalTasks; i++) {
- ZkCoordinator coordinator = new ZkCoordinator(dynamicPartitionConnections, stormConf, spoutConfig, state, i, totalTasks, "test-id", reader);
+ ZkCoordinator coordinator = new ZkCoordinator(dynamicPartitionConnections, partitionStateManagerFactory, stormConf, spoutConfig, i, totalTasks, "test-id", reader);
coordinatorList.add(coordinator);
}
return coordinatorList;