Fix bug when get stopped zk state or env.

Author: tinawenqiao

docs/definitions.adoc

@@ -380,7 +380,7 @@ __optional__|integer(int32)
 |**minLatency** +
 __optional__|integer(int32)
 |**mode** +
-__optional__|enum (Leader, Follower, Observer)
+__optional__|enum (Leader, Follower, Observer, Standalone)
 |**nodes** +
 __optional__|integer(int32)
 |**outstanding** +

docs/index.html

@@ -5393,7 +5393,7 @@ <h3 id="_zkserverstat"><a class="anchor" href="#_zkserverstat"></a>5.24. ZkServe
 <tr>
 <td class="tableblock halign-left valign-middle"><p class="tableblock"><strong>mode</strong><br>
 <em>optional</em></p></td>
-<td class="tableblock halign-left valign-middle"><p class="tableblock">enum (Leader, Follower, Observer)</p></td>
+<td class="tableblock halign-left valign-middle"><p class="tableblock">enum (Leader, Follower, Observer, Standalone)</p></td>
 </tr>
 <tr>
 <td class="tableblock halign-left valign-middle"><p class="tableblock"><strong>nodes</strong><br>
@@ -5433,7 +5433,7 @@ <h3 id="_zkserverstat"><a class="anchor" href="#_zkserverstat"></a>5.24. ZkServe
 </div>
 <div id="footer">
 <div id="footer-text">
-Last updated 2018-02-10 00:44:08 CST
+Last updated 2018-11-13 00:36:04 +08:00
 </div>
 </div>
 </body>

docs/index.pdf

@@ -98,7 +98,7 @@
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-log4j</artifactId>
-            <version>RELEASE</version>
+            <version>1.3.8.RELEASE</version>
         </dependency>
 
         <dependency>

pom.xml

@@ -3,5 +3,6 @@
 public enum ZkServerMode {
   Leader,
   Follower,
-  Observer
+  Observer,
+  Standalone
 }

src/docs/swagger/swagger.json

@@ -1,8 +1,11 @@
 package org.gnuhpc.bigdata.service;
 
 import com.google.common.net.HostAndPort;
+import java.util.HashMap;
+import java.util.List;
 import lombok.extern.log4j.Log4j;
 import org.gnuhpc.bigdata.constant.ZkServerCommand;
+import org.gnuhpc.bigdata.exception.ServiceNotAvailableException;
 import org.gnuhpc.bigdata.model.ZkServerEnvironment;
 import org.gnuhpc.bigdata.model.ZkServerStat;
 import org.gnuhpc.bigdata.utils.ZookeeperUtils;
@@ -15,36 +18,39 @@
 @Service
 @Log4j
 public class ZookeeperService {
-    @Autowired
-    private ZookeeperUtils zookeeperUtils;
 
-    public Map<HostAndPort, ZkServerStat> stat() {
-        return zookeeperUtils.getZookeeperConfig().getHostAndPort().stream()
-                .collect(Collectors.toMap(
-                        hp -> hp,
-                        hp -> zookeeperUtils.parseStatResult(
-                                zookeeperUtils.executeCommand(
-                                        hp.getHostText(),
-                                        hp.getPort(),
-                                        ZkServerCommand.stat.toString()
-                                )
-                        )
-                ));
-    }
+  @Autowired
+  private ZookeeperUtils zookeeperUtils;
 
-    public Map<HostAndPort, ZkServerEnvironment> environment() {
-        return zookeeperUtils.getZookeeperConfig().getHostAndPort().stream()
-                .collect(Collectors.toMap(
-                        hp -> hp,
-                        hp -> zookeeperUtils.parseEnvResult(
-                                zookeeperUtils.executeCommand(
-                                        hp.getHostText(),
-                                        hp.getPort(),
-                                        ZkServerCommand.envi.toString()
-                                )
-                        )
-                ));
+  public Map<HostAndPort, ZkServerStat> stat() {
+    List<HostAndPort> hostAndPortList = zookeeperUtils.getZookeeperConfig().getHostAndPort();
+    Map<HostAndPort, ZkServerStat> result = new HashMap<>();
+    for (int i = 0; i < hostAndPortList.size(); i++) {
+      HostAndPort hp = hostAndPortList.get(i);
+      try {
+        result.put(hp, zookeeperUtils.parseStatResult(zookeeperUtils
+            .executeCommand(hp.getHostText(), hp.getPort(), ZkServerCommand.stat.toString())));
+      } catch (ServiceNotAvailableException serviceNotAvailbleException) {
+        log.warn("Execute " + ZkServerCommand.stat.toString() + " command failed. Exception:"
+            + serviceNotAvailbleException);
+      }
     }
+    return result;
+  }
 
-
+  public Map<HostAndPort, ZkServerEnvironment> environment() {
+    List<HostAndPort> hostAndPortList = zookeeperUtils.getZookeeperConfig().getHostAndPort();
+    Map<HostAndPort, ZkServerEnvironment> result = new HashMap<>();
+    for (int i = 0; i < hostAndPortList.size(); i++) {
+      HostAndPort hp = hostAndPortList.get(i);
+      try {
+        result.put(hp, zookeeperUtils.parseEnvResult(zookeeperUtils
+            .executeCommand(hp.getHostText(), hp.getPort(), ZkServerCommand.envi.toString())));
+      } catch (ServiceNotAvailableException serviceNotAvailbleException) {
+        log.warn("Execute " + ZkServerCommand.envi.toString() + " command failed. Exception:"
+            + serviceNotAvailbleException);
+      }
+    }
+    return result;
+  }
 }

src/main/java/org/gnuhpc/bigdata/constant/ZkServerMode.java

@@ -74,7 +74,7 @@ public KafkaConsumer createNewConsumer(String consumerGroup) {
         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
         properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
         properties.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "100000000");
-        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");
+        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5");
         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                 StringDeserializer.class.getCanonicalName());
         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,

src/main/java/org/gnuhpc/bigdata/service/ZookeeperService.java

@@ -1,5 +1,5 @@
 kafka:
-  brokers: localhost:9092,localhost:9093,localhost:9094
+  brokers: localhost:19092,localhost:19093,localhost:19095
   offset:
     topic: "__consumer_offsets"
     partitions: 50
@@ -8,7 +8,7 @@ kafka:
     topic: "health"
 
 zookeeper:
-  uris: 127.0.0.1:2181
+  uris: 127.0.0.1:2183,127.0.0.1:2182
 
 jmx:
   kafka: