Modify readme.

Author: tinawenqiao

README.adoc

@@ -8,13 +8,17 @@ Kafka/ZK REST API is to provide the production-ready endpoints to perform some a
 
 // tag::base-t[]
 .Following features are provided:
+* cluster/nodes/controller info describe
 * Broker List
-* Topic create/delete/describe
-* Topic config create/update/delete/list
-* Topic partition add and reassign
-* Consumer group(old zookeeper based/new kafka based) list/describe
-* Offset check/reset
+* Broker config get/update, dynamic config get/update/delete
+* Log dirs describe, filtered by brokers/topic/partition/logdirs
+* Topic create/delete/describe/list
+* Topic config create/update/list
+* Topic partition add and reassign, replicas movement between log directories is supported
+* Consumer group list/describe/delete
 * Consumer Group Lag check
+* Offset reset by earliest/latest/timestamp
+* Contents view of message by different decoder, avro is also supported
 * Collect JMX metrics from brokers that expose JMX metrics +
   More details refer to https://github.com/gnuhpc/Kafka-zk-restapi/blob/master/docs/JMXCollector.adoc[JMXCollector API Specification]
 * Secure the REST API with Spring Security
@@ -82,7 +86,7 @@ Step 3: Use user controller API to add user to security file security/security.y
 * No need to restart server after adding new user or update user info. Timing thread introduced in Step 1 will refresh the user list according to your settings.
 
 === Support Kafka Version Information
-Currently, this rest api (master branch) supports Kafka 0.10.x brokers. The master branch is the most active branch. We're going to get down to the work of supporting the Kafka 1.x version.
+Currently, this rest api (master branch) supports Kafka 1.1.1 brokers. The master branch is the most active branch.
 
 *For 0.11.x, please checkout the branch 0.11.x by calling the command:*
 
@@ -96,8 +100,7 @@ __BasePath__ : /
 You can access Swagger-UI by accessing http://127.0.0.1:8121/api
 
 
-=== https://github.com/gnuhpc/Kafka-zk-restapi/blob/master/docs/paths.adoc[API LIST for 0.10]
-=== https://github.com/gnuhpc/Kafka-zk-restapi/blob/0.11.x/docs/paths.adoc[API LIST for 0.11.x]
+=== https://github.com/gnuhpc/Kafka-zk-restapi/blob/master/docs/paths.adoc[API LIST for 1.1.1]
 
 
 * kafka-controller : Kafka Api
@@ -106,13 +109,12 @@ You can access Swagger-UI by accessing http://127.0.0.1:8121/api
 * user-controller : User management Api
 
 
-=== https://github.com/gnuhpc/Kafka-zk-restapi/blob/master/docs/definitions.adoc[Data Model Definitions for 0.10]
-=== https://github.com/gnuhpc/Kafka-zk-restapi/blob/0.11.x/docs/definitions.adoc[Data Model Definitions for 0.11.x]
+=== https://github.com/gnuhpc/Kafka-zk-restapi/blob/master/docs/definitions.adoc[Data Model Definitions for 1.1.1]
 
 
 === Version information
 [%hardbreaks]
-__Version__ : 0.1.0
+__Version__ : 1.1.1
 
 
 === Contact information

pics/ShowApi.png

@@ -50,13 +50,16 @@
 import org.springframework.web.bind.annotation.ResponseStatus;
 import org.springframework.web.bind.annotation.RestController;
 
-/** Created by gnuhpc on 2017/7/16. */
+/**
+ * Created by gnuhpc on 2017/7/16.
+ */
 @Log4j
 @RequestMapping("/kafka")
 @RestController
 public class KafkaController {
 
-  @Autowired private KafkaAdminService kafkaAdminService;
+  @Autowired
+  private KafkaAdminService kafkaAdminService;
 
 //  @Autowired private KafkaProducerService kafkaProducerService;
 
@@ -91,15 +94,16 @@ public Map<Integer, Map<String, LogDirInfo>> describeLogDirs(
       @RequestParam(required = false) List<Integer> brokerList,
       @RequestParam(required = false) List<String> logDirList,
       @RequestBody(required = false) Map<String, List<Integer>> topicPartitionMap) {
-    return kafkaAdminService.describeLogDirsByBrokerAndTopic(brokerList, logDirList, topicPartitionMap);
+    return kafkaAdminService
+        .describeLogDirsByBrokerAndTopic(brokerList, logDirList, topicPartitionMap);
   }
 
-
-  @GetMapping(value = "/brokers/replicalogdirs")
-  @ApiOperation(value = "Describe replica log dirs.")
-  public Map<TopicPartitionReplica, ReplicaLogDirInfo> describeReplicaLogDirs(
-      @RequestParam List<TopicPartitionReplica> replicas) {
-    return kafkaAdminService.describeReplicaLogDirs(replicas);
+  @GetMapping(value = "/brokers/replicalogdir/{brokerId}/{topic}/{partition}")
+  @ApiOperation(value = "Describe replica log dir.")
+  public ReplicaLogDirInfo describeReplicaLogDirs(@PathVariable int brokerId,
+      @PathVariable String topic, @PathVariable int partition) {
+    TopicPartitionReplica replica = new TopicPartitionReplica(topic, partition, brokerId);
+    return kafkaAdminService.describeReplicaLogDir(replica);
   }
 
   @GetMapping(value = "/brokers/{brokerId}/conf")
@@ -179,7 +183,8 @@ public List<Record> getMessage(
       @RequestParam(required = false, defaultValue = "StringDeserializer") String keyDecoder,
       @RequestParam(required = false, defaultValue = "StringDeserializer") String valueDecoder,
       @RequestParam(required = false) String avroSchema,
-      @RequestParam(required = false, defaultValue = "30000") long fetchTimeoutMs) throws ApiException {
+      @RequestParam(required = false, defaultValue = "30000") long fetchTimeoutMs)
+      throws ApiException {
     return kafkaAdminService.getRecordsByOffset(topic, partition, offset, maxRecords, keyDecoder,
         valueDecoder, avroSchema, fetchTimeoutMs);
   }
@@ -241,7 +246,8 @@ public Map<String, GeneralResponse> addPartition(@RequestBody List<AddPartition>
 
   @PostMapping(value = "/partitions/reassign/generate")
   @ApiOperation(value = "Generate plan for the partition reassignment")
-  public List<ReassignModel> generateReassignPartitions(@RequestBody ReassignWrapper reassignWrapper) {
+  public List<ReassignModel> generateReassignPartitions(
+      @RequestBody ReassignWrapper reassignWrapper) {
     return kafkaAdminService.generateReassignPartition(reassignWrapper);
   }
 
@@ -260,9 +266,9 @@ public ReassignStatus executeReassignPartitions(
   @ApiOperation(value = "Check the partition reassignment process")
   @ApiResponses(
       value = {
-        @ApiResponse(code = 1, message = "Reassignment Completed"),
-        @ApiResponse(code = 0, message = "Reassignment In Progress"),
-        @ApiResponse(code = -1, message = "Reassignment Failed")
+          @ApiResponse(code = 1, message = "Reassignment Completed"),
+          @ApiResponse(code = 0, message = "Reassignment In Progress"),
+          @ApiResponse(code = -1, message = "Reassignment Failed")
       })
   public ReassignStatus checkReassignPartitions(@RequestBody ReassignModel reassign) {
     return kafkaAdminService.checkReassignStatus(reassign);
@@ -314,7 +320,7 @@ public ConsumerGroupMeta getConsumerGroupMeta(@PathVariable String consumerGroup
   public List<ConsumerGroupMeta> getConsumerGroupsMeta() {
     Set<String> consumerGroupList = kafkaAdminService.listAllNewConsumerGroups();
     List<ConsumerGroupMeta> consumerGroupMetaList = new ArrayList<>();
-    for (String consumerGroup:consumerGroupList) {
+    for (String consumerGroup : consumerGroupList) {
       if (kafkaAdminService.isNewConsumerGroup(consumerGroup)) {
         consumerGroupMetaList.add(kafkaAdminService.getConsumerGroupMeta(consumerGroup));
       } else {
@@ -369,10 +375,10 @@ public GeneralResponse resetOffset(
       @PathVariable int partition,
       @PathVariable String consumergroup,
       @PathVariable
-          @ApiParam(
-              value =
-                  "[earliest/latest/{long}/yyyy-MM-dd HH:mm:ss.SSS] can be supported. "
-                      + "The date type is only valid for new consumer group.")
+      @ApiParam(
+          value =
+              "[earliest/latest/{long}/yyyy-MM-dd HH:mm:ss.SSS] can be supported. "
+                  + "The date type is only valid for new consumer group.")
           String offset,
       @PathVariable ConsumerType type) {
     return kafkaAdminService.resetOffset(topic, partition, consumergroup, type, offset);

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

@@ -536,6 +536,13 @@ public Map<Integer, Map<String, LogDirInfo>> describeLogDirsByBrokerAndTopic(
     return logDirInfosByBroker;
   }
 
+  public ReplicaLogDirInfo describeReplicaLogDir(TopicPartitionReplica topicPartitionReplica) {
+    Map<TopicPartitionReplica, ReplicaLogDirInfo> replicaLogDirInfoMap =
+        describeReplicaLogDirs(Collections.singletonList(topicPartitionReplica));
+
+    return replicaLogDirInfoMap.get(topicPartitionReplica);
+  }
+
   public Map<TopicPartitionReplica, ReplicaLogDirInfo> describeReplicaLogDirs(
       List<TopicPartitionReplica> replicas) {
     org.apache.kafka.clients.admin.AdminClient kafkaAdminClient = createKafkaAdminClient();