Modify consumer poll timeout in KafkaAdminService.getMessage and add some log info.Add lspath and getnode path in zookeeper utils.

Author: tinawenqiao

docs/index.pdf

@@ -7,8 +7,12 @@
 import org.gnuhpc.bigdata.model.ZkServerStat;
 import org.gnuhpc.bigdata.service.ZookeeperService;
 import org.gnuhpc.bigdata.utils.ZookeeperUtils;
+import org.gnuhpc.bigdata.validator.ZKNodePathExistConstraint;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.*;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
 
 import java.util.List;
 import java.util.Map;
@@ -26,16 +30,25 @@ public class ZookeeperController {
     @Autowired
     private ZookeeperService zookeeperService;
 
-    @GetMapping("/ls/{path}")
+    @GetMapping("/ls/path")
     @ApiOperation(value = "List a zookeeper path")
-    public List<String> ls(@PathVariable("path") String path){
+    public List<String> ls(@RequestParam String path){
+        return zookeeperUtils.lsPath(path);
+        /*
         try {
-            return zookeeperUtils.getCuratorClient().getChildren().forPath("/"+path);
+            return zookeeperUtils.getCuratorClient().getChildren().forPath(path);
         } catch (Exception e) {
             e.printStackTrace();
         }
 
         return null;
+        */
+    }
+
+    @GetMapping("/get/path")
+    @ApiOperation(value = "Get data of a zookeeper path")
+    public Map<String, String> get(@RequestParam String path){
+        return zookeeperUtils.getNodeData(path);
     }
 
     @GetMapping("/connstate")

src/main/java/org/gnuhpc/bigdata/controller/ZookeeperController.java

@@ -626,7 +626,7 @@ private List<ConsumerGroupDesc> setNewCGD(String consumerGroup, String topic) {
 
 
     public String getMessage(@TopicExistConstraint String topic, int partition, long offset, String decoder, String avroSchema) {
-        KafkaConsumer consumer = kafkaUtils.createNewConsumer();
+        KafkaConsumer consumer = kafkaUtils.createNewConsumer(String.valueOf(System.currentTimeMillis()));
         TopicPartition tp = new TopicPartition(topic, partition);
         long beginningOffset = getBeginningOffset(topic, partition);
         long endOffset = getEndOffset(topic, partition);
@@ -647,18 +647,21 @@ public String getMessage(@TopicExistConstraint String topic, int partition, long
 
         String last = null;
 
-        ConsumerRecords<String, String> crs = consumer.poll(channelRetryBackoffMs);
+        //ConsumerRecords<String, String> crs = consumer.poll(channelRetryBackoffMs);
+        ConsumerRecords<String, String> crs = consumer.poll(3000);
+        log.info("Seek to offset:" + offset + ", topic:" + topic + ", partition:" + partition + ", crs.count:" + crs.count());
         if (crs.count() != 0) {
             Iterator<ConsumerRecord<String, String>> it = crs.iterator();
             while (it.hasNext()) {
                 ConsumerRecord<String, String> initCr = it.next();
                 last = "Value: " + initCr.value() + ", Offset: " + String.valueOf(initCr.offset());
+                log.info("Value: " + initCr.value() + ", initCr.Offset: " + String.valueOf(initCr.offset()));
                 if (last != null && initCr.offset() == offset) {
                     break;
                 }
             }
         }
-
+        log.info("last:" + last);
         consumer.close();
         return last;
     }

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

@@ -1,5 +1,6 @@
 package org.gnuhpc.bigdata.utils;
 
+import com.google.common.base.Charsets;
 import kafka.utils.ZKStringSerializer$;
 import kafka.utils.ZkUtils;
 import lombok.Getter;
@@ -12,23 +13,24 @@
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.GetDataBuilder;
 import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.utils.ZKPaths;
+import org.apache.zookeeper.data.Stat;
 import org.gnuhpc.bigdata.config.ZookeeperConfig;
 import org.gnuhpc.bigdata.constant.ZkServerMode;
 import org.gnuhpc.bigdata.exception.ServiceNotAvailableException;
 import org.gnuhpc.bigdata.model.ZkServerClient;
 import org.gnuhpc.bigdata.model.ZkServerEnvironment;
 import org.gnuhpc.bigdata.model.ZkServerStat;
+import org.gnuhpc.bigdata.validator.ZKNodePathExistConstraint;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Configuration;
+import org.springframework.validation.annotation.Validated;
 
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.Socket;
-import java.net.UnknownHostException;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
+import java.util.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -39,6 +41,7 @@
 @Log4j
 @Setter
 @Getter
+@Validated
 public class ZookeeperUtils {
 
     //For Stat Command parse
@@ -233,5 +236,47 @@ public ZkServerEnvironment parseEnvResult(final List<String> result) {
         return environment;
     }
 
+    public List<String> lsPath(@ZKNodePathExistConstraint String path) {
+        try {
+            return curatorClient.getChildren().forPath(path);
+        } catch (Exception e) {
+            log.error("ls path fail! path: " + path + ", error: {}" + e);
+            return null;
+        }
+    }
+
+    public Map<String, String> getNodeData(@ZKNodePathExistConstraint String path) {
+        Map<String, String> map = new HashMap<>();
+
+        try {
+            List<String> childrens = curatorClient.getChildren().forPath(path);
+            GetDataBuilder dataBuilder = curatorClient.getData();
+            if (childrens != null && childrens.size() > 0) {
+                for (int i = 0; i < childrens.size(); i++) {
+                    String child = childrens.get(i);
+                    String childPath = ZKPaths.makePath(path, child);
+                    byte[] bytes = dataBuilder.forPath(childPath);
+                    map.put(childPath, (bytes!=null)?(new String(bytes, Charsets.UTF_8)):(null));
+                }
+            } else {
+                byte[] bytes = dataBuilder.forPath(path);
+                map.put(path, (bytes!=null)?(new String(bytes, Charsets.UTF_8)):(null));
+            }
+
+        } catch (Exception e) {
+            log.error("get node data fail! path: " + path + ", error: {}" + e);
+        }
+
+        return map;
+    }
+
+    public Stat getNodePathStat(String path) {
+        try {
+            return curatorClient.checkExists().forPath(path);
+        } catch (Exception e) {
+            log.error("get node data fail! path: " + path + ", error: {}" + e);
+        }
+        return null;
+    }
 }
 

src/main/java/org/gnuhpc/bigdata/utils/ZookeeperUtils.java

@@ -0,0 +1,19 @@
+package org.gnuhpc.bigdata.validator;
+
+import javax.validation.Constraint;
+import javax.validation.Payload;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Constraint(validatedBy = ZKNodePathExistValidator.class)
+@Target( { ElementType.METHOD, ElementType.FIELD, ElementType.PARAMETER})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface ZKNodePathExistConstraint {
+  String message() default "Non-exist ZooKeeper Node path!";
+
+  Class<?>[] groups() default {};
+
+  Class<? extends Payload>[] payload() default {};
+}

src/main/java/org/gnuhpc/bigdata/validator/ZKNodePathExistConstraint.java

@@ -0,0 +1,19 @@
+package org.gnuhpc.bigdata.validator;
+
+import org.gnuhpc.bigdata.utils.ZookeeperUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import javax.validation.ConstraintValidator;
+import javax.validation.ConstraintValidatorContext;
+
+public class ZKNodePathExistValidator implements ConstraintValidator<ZKNodePathExistConstraint, String> {
+  @Autowired
+  private ZookeeperUtils zookeeperUtils;
+
+  public void initialize(ZKNodePathExistConstraint constraint) {
+  }
+
+  public boolean isValid(String path, ConstraintValidatorContext context) {
+    return (zookeeperUtils.getNodePathStat(path)!=null);
+  }
+}