Repository: strimzi/strimzi-kafka-bridge (java)

Issue:
Adding operation for getting current subscribed topics
On the `/consumers/{groupid}/instances/{name}/subscription` endpoint should be possible to make a GET (HTTP request) in order to get the currently subscribed topics for the consumer.
It's not an administrative API because it's provided by the underlying native Kafka consumer client via the `assignment` method (which provides topics with related partitions).
The GET should Accept `application/vnd.kafka.v2+json` and the bridge should return something like:

```shell
HTTP/1.1 200 OK
Content-Type: application/vnd.kafka.v2+json

{
  "topics": [
    "my-topic1",
    "my-topic2"
  ]
}
```

Pull request description:
Add a way how to list subscribed topics
Fixes #298 

Created a topic with 5 partitions, subscribed a consumer via the bridge, then added a new consumer via console.
Before
```
{
    "topics": [
        "test"
    ],
    "partitions": [
        {
            "test": [
                1,
                0,
                3,
                2,
                4
            ]
        }
    ]
}
```

After
```
{
    "topics": [
        "test"
    ],
    "partitions": [
        {
            "test": [
                3,
                4
            ]
        }
    ]
}
```

Patch:
diff --git a/documentation/book/api/definitions.adoc b/documentation/book/api/definitions.adoc
index 476ae05..b2ea0b2 100644
--- a/documentation/book/api/definitions.adoc
+++ b/documentation/book/api/definitions.adoc
@@ -2,6 +2,11 @@
 [[_definitions]]
 == Definitions
 
+[[_assignedtopicpartitions]]
+=== AssignedTopicPartitions
+__Type__ : < string, < integer (int32) > array > map
+
+
 [[_consumer]]
 === Consumer
 
@@ -187,6 +192,19 @@ __optional__|< <<_producerrecordtopartition,ProducerRecordToPartition>> > array
 |===
 
 
+[[_subscribedtopiclist]]
+=== SubscribedTopicList
+
+[options="header", cols=".^3,.^4"]
+|===
+|Name|Schema
+|**partitions** +
+__optional__|< <<_assignedtopicpartitions,AssignedTopicPartitions>> > array
+|**topics** +
+__optional__|<<_topics,Topics>>
+|===
+
+
 [[_topics]]
 === Topics
 
diff --git a/documentation/book/api/paths.adoc b/documentation/book/api/paths.adoc
index d5e312d..6fba4db 100644
--- a/documentation/book/api/paths.adoc
+++ b/documentation/book/api/paths.adoc
@@ -770,6 +770,73 @@ __required__|List of topics to which the consumer will subscribe.|<<_topics,Topi
 ----
 
 
+[[_listsubscriptions]]
+=== GET /consumers/{groupid}/instances/{name}/subscription
+
+==== Description
+Retrieves a list of the topics to which the consumer is subscribed.
+
+
+==== Parameters
+
+[options="header", cols=".^2,.^3,.^9,.^4"]
+|===
+|Type|Name|Description|Schema
+|**Path**|**groupid** +
+__required__|ID of the consumer group to which the subscribed consumer belongs.|string
+|**Path**|**name** +
+__required__|Name of the consumer that you want to unsubscribe from topics.|string
+|===
+
+
+==== Responses
+
+[options="header", cols=".^2,.^14,.^4"]
+|===
+|HTTP Code|Description|Schema
+|**200**|List of subscribed topics and partitions.|<<_subscribedtopiclist,SubscribedTopicList>>
+|**404**|The specified consumer instance was not found.|<<_error,Error>>
+|===
+
+
+==== Produces
+
+* `application/vnd.kafka.v2+json`
+
+
+==== Tags
+
+* Consumers
+
+
+==== Example HTTP response
+
+===== Response 200
+[source,json]
+----
+{
+  "topics" : [ "my-topic1", "my-topic2" ],
+  "partitions" : [ {
+    "my-topic1" : [ 1, 2, 3 ]
+  }, {
+    "my-topic2" : [ 1 ]
+  } ]
+}
+----
+
+
+===== Response 404
+[source,json]
+----
+{
+  "application/vnd.kafka.v2+json" : {
+    "error_code" : 404,
+    "message" : "The specified consumer instance was not found."
+  }
+}
+----
+
+
 [[_unsubscribe]]
 === DELETE /consumers/{groupid}/instances/{name}/subscription
 
diff --git a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
index 4832436..c92440b 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
@@ -194,6 +194,14 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
         this.consumer.unsubscribe(this::unsubscribeHandler);
     }
 
+    /**
+     * Returns all the topics which the consumer currently subscribes
+     */
+    protected void listSubscriptions(Handler<AsyncResult<Set<TopicPartition>>> handler) {
+        log.info("Listing subscribed topics {}", this.topicSubscriptions);
+        this.consumer.assignment(handler);
+    }
+
     /**
      * Subscribe to topics via the provided pattern represented by a Java regex
      *
@@ -573,7 +581,6 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
         this.unsubscribeHandler = handler;
     }
 
-
     /**
      * Set the handler called after a request for a specific partition is executed
      *
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
index df8965d..3f06211 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -128,6 +128,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
                 routerFactory.addHandlerByOperationId(this.DELETE_CONSUMER.getOperationId().toString(), this.DELETE_CONSUMER);
                 routerFactory.addHandlerByOperationId(this.SUBSCRIBE.getOperationId().toString(), this.SUBSCRIBE);
                 routerFactory.addHandlerByOperationId(this.UNSUBSCRIBE.getOperationId().toString(), this.UNSUBSCRIBE);
+                routerFactory.addHandlerByOperationId(this.LIST_SUBSCRIPTIONS.getOperationId().toString(), this.LIST_SUBSCRIPTIONS);
                 routerFactory.addHandlerByOperationId(this.ASSIGN.getOperationId().toString(), this.ASSIGN);
                 routerFactory.addHandlerByOperationId(this.POLL.getOperationId().toString(), this.POLL);
                 routerFactory.addHandlerByOperationId(this.COMMIT.getOperationId().toString(), this.COMMIT);
@@ -263,6 +264,11 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         processConsumer(routingContext);
     }
 
+    private void listSubscriptions(RoutingContext routingContext) {
+        this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.LIST_SUBSCRIPTIONS);
+        processConsumer(routingContext);
+    }
+
     private void assign(RoutingContext routingContext) {
         this.httpBridgeContext.setOpenApiOperation(HttpOpenApiOperations.ASSIGN);
         processConsumer(routingContext);
@@ -506,6 +512,14 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         }
     };
 
+    HttpOpenApiOperation LIST_SUBSCRIPTIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_SUBSCRIPTIONS) {
+
+        @Override
+        public void process(RoutingContext routingContext) {
+            listSubscriptions(routingContext);
+        }
+    };
+
     HttpOpenApiOperation ASSIGN = new HttpOpenApiOperation(HttpOpenApiOperations.ASSIGN) {
     
         @Override
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java
index 9cca81f..54bdef2 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperations.java
@@ -16,6 +16,7 @@ public enum HttpOpenApiOperations {
     DELETE_CONSUMER("deleteConsumer"),
     SUBSCRIBE("subscribe"),
     UNSUBSCRIBE("unsubscribe"),
+    LIST_SUBSCRIPTIONS("listSubscriptions"),
     ASSIGN("assign"),
     POLL("poll"),
     COMMIT("commit"),
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
index 21c0c16..5feef51 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -345,6 +345,44 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         }
     }
 
+    public void doListSubscriptions(RoutingContext routingContext) {
+        this.listSubscriptions(listSubscriptionsResult -> {
+
+            if (listSubscriptionsResult.succeeded()) {
+                JsonObject root = new JsonObject();
+                JsonArray topicsArray = new JsonArray();
+                JsonArray partitionsArray = new JsonArray();
+
+                HashMap<String, JsonArray> partitions = new HashMap<>();
+                for (TopicPartition topicPartition: listSubscriptionsResult.result()) {
+                    if (!topicsArray.contains(topicPartition.getTopic())) {
+                        topicsArray.add(topicPartition.getTopic());
+                    }
+                    if (!partitions.containsKey(topicPartition.getTopic())) {
+                        partitions.put(topicPartition.getTopic(), new JsonArray());
+                    }
+                    partitions.put(topicPartition.getTopic(), partitions.get(topicPartition.getTopic()).add(topicPartition.getPartition()));
+                }
+                for (Map.Entry<String, JsonArray> part: partitions.entrySet()) {
+                    JsonObject topic = new JsonObject();
+                    topic.put(part.getKey(), part.getValue());
+                    partitionsArray.add(topic);
+                }
+                root.put("topics", topicsArray);
+                root.put("partitions", partitionsArray);
+
+                HttpUtils.sendResponse(routingContext, HttpResponseStatus.OK.code(), BridgeContentType.KAFKA_JSON, root.toBuffer());
+            } else {
+                HttpBridgeError error = new HttpBridgeError(
+                        HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                        listSubscriptionsResult.cause().getMessage()
+                );
+                HttpUtils.sendResponse(routingContext, HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
+                        BridgeContentType.KAFKA_JSON, error.toJson().toBuffer());
+            }
+        });
+    }
+
     public void doUnsubscribe(RoutingContext routingContext) {
         this.setUnsubscribeHandler(unsubscribeResult -> {
             if (unsubscribeResult.succeeded()) {
@@ -426,6 +464,9 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
             case UNSUBSCRIBE:
                 doUnsubscribe(routingContext);
                 break;
+            case LIST_SUBSCRIPTIONS:
+                doListSubscriptions(routingContext);
+                break;
 
             default:
                 throw new IllegalArgumentException("Unknown Operation: " + this.httpBridgeContext.getOpenApiOperation());
diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
index 7374835..2ff94db 100644
--- a/src/main/resources/openapi.json
+++ b/src/main/resources/openapi.json
@@ -235,6 +235,43 @@
             ]
         },
         "/consumers/{groupid}/instances/{name}/subscription": {
+            "get": {
+                "tags": [
+                    "Consumers"
+                ],
+                "responses": {
+                    "200": {
+                        "description": "List of subscribed topics and partitions.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/SubscribedTopicList"
+                                }
+                            }
+                        }
+                    },
+                    "404": {
+                        "description": "The specified consumer instance was not found.",
+                        "content": {
+                            "application/vnd.kafka.v2+json": {
+                                "schema": {
+                                    "$ref": "#/components/schemas/Error"
+                                },
+                                "examples": {
+                                    "response": {
+                                        "value": {
+                                            "error_code": 404,
+                                            "message": "The specified consumer instance was not found."
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                },
+                "operationId": "listSubscriptions",
+                "description": "Retrieves a list of the topics to which the consumer is subscribed."
+            },
             "post": {
                 "tags": [
                     "Consumers"
@@ -1473,6 +1510,59 @@
                         "topic2"
                     ]
                 }
+            },
+            "AssignedTopicPartitions": {
+                "title": "AssignedTopicPartitions",
+                "type": "object",
+                "additionalProperties": {
+                    "type": "array",
+                    "items": {
+                        "format": "int32",
+                        "type": "integer"
+                    }
+                },
+                "example": {
+                    "my-topic1": [
+                        1,
+                        2,
+                        3
+                    ]
+                }
+            },
+            "SubscribedTopicList": {
+                "title": "SubscribedTopicList",
+                "type": "object",
+                "properties": {
+                    "topics": {
+                        "$ref": "#/components/schemas/Topics"
+                    },
+                    "partitions": {
+                        "type": "array",
+                        "items": {
+                            "$ref": "#/components/schemas/AssignedTopicPartitions"
+                        }
+                    }
+                },
+                "example": {
+                    "topics": [
+                        "my-topic1",
+                        "my-topic2"
+                    ],
+                    "partitions": [
+                        {
+                            "my-topic1": [
+                                1,
+                                2,
+                                3
+                            ]
+                        },
+                        {
+                            "my-topic2": [
+                                1
+                            ]
+                        }
+                    ]
+                }
             }
         }
     },
diff --git a/src/main/resources/openapiv2.json b/src/main/resources/openapiv2.json
index 43af72e..804b0c2 100644
--- a/src/main/resources/openapiv2.json
+++ b/src/main/resources/openapiv2.json
@@ -219,6 +219,36 @@
       ]
     },
     "/consumers/{groupid}/instances/{name}/subscription": {
+      "get": {
+        "produces": [
+          "application/vnd.kafka.v2+json"
+        ],
+        "tags": [
+          "Consumers"
+        ],
+        "responses": {
+          "200": {
+            "description": "List of subscribed topics and partitions.",
+            "schema": {
+              "$ref": "#/definitions/SubscribedTopicList"
+            }
+          },
+          "404": {
+            "description": "The specified consumer instance was not found.",
+            "schema": {
+              "$ref": "#/definitions/Error"
+            },
+            "examples": {
+              "application/vnd.kafka.v2+json": {
+                "error_code": 404,
+                "message": "The specified consumer instance was not found."
+              }
+            }
+          }
+        },
+        "operationId": "listSubscriptions",
+        "description": "Retrieves a list of the topics to which the consumer is subscribed."
+      },
       "post": {
         "tags": [
           "Consumers"
@@ -1287,6 +1317,59 @@
           "topic2"
         ]
       }
+    },
+    "AssignedTopicPartitions": {
+      "title": "AssignedTopicPartitions",
+      "type": "object",
+      "additionalProperties": {
+        "type": "array",
+        "items": {
+          "format": "int32",
+          "type": "integer"
+        }
+      },
+      "example": {
+        "my-topic1": [
+          1,
+          2,
+          3
+        ]
+      }
+    },
+    "SubscribedTopicList": {
+      "title": "SubscribedTopicList",
+      "type": "object",
+      "properties": {
+        "topics": {
+          "$ref": "#/definitions/Topics"
+        },
+        "partitions": {
+          "type": "array",
+          "items": {
+            "$ref": "#/definitions/AssignedTopicPartitions"
+          }
+        }
+      },
+      "example": {
+        "topics": [
+          "my-topic1",
+          "my-topic2"
+        ],
+        "partitions": [
+          {
+            "my-topic1": [
+              1,
+              2,
+              3
+            ]
+          },
+          {
+            "my-topic2": [
+              1
+            ]
+          }
+        ]
+      }
     }
   },
   "tags": [


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java
index 27e8ad1..d419346 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerSubscriptionTest.java
@@ -5,6 +5,7 @@
 package io.strimzi.kafka.bridge.http;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
+import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.strimzi.kafka.bridge.utils.Urls;
 import io.vertx.core.json.JsonArray;
@@ -156,4 +157,62 @@ public class ConsumerSubscriptionTest extends HttpBridgeTestBase {
         subscribe.get(TEST_TIMEOUT, TimeUnit.SECONDS);
         context.completeNow();
     }
+
+    @Test
+    void listConsumerSubscriptions(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
+        String topic = "listConsumerSubscriptions";
+        String topic2 = "listConsumerSubscriptions2";
+        kafkaCluster.createTopic(topic, 1, 1);
+        kafkaCluster.createTopic(topic2, 4, 1);
+
+        String name = "my-kafka-consumer";
+        String groupId = "my-group";
+
+        JsonObject json = new JsonObject();
+        json.put("name", name);
+        json.put("format", "json");
+
+        JsonArray topics = new JsonArray();
+        topics.add(topic);
+        topics.add(topic2);
+
+        JsonObject topicsRoot = new JsonObject();
+        topicsRoot.put("topics", topics);
+
+        consumerService()
+                .createConsumer(context, groupId, json)
+                .subscribeConsumer(context, groupId, name, topic)
+                .subscribeConsumer(context, groupId, name, topic2);
+
+        CompletableFuture<Boolean> consume = new CompletableFuture<>();
+
+        // poll to subscribe
+        consumerService()
+                .consumeRecordsRequest(groupId, name, BridgeContentType.KAFKA_JSON_JSON)
+                .as(BodyCodec.jsonObject())
+                .send(ar -> consume.complete(true));
+
+        CompletableFuture<Boolean> listSubscriptions = new CompletableFuture<>();
+        consumerService()
+                .listSubscriptionsConsumerRequest(groupId, name)
+                .as(BodyCodec.jsonObject())
+                .send(ar -> {
+                    context.verify(() -> {
+                        assertTrue(ar.succeeded());
+
+                        HttpResponse<JsonObject> response = ar.result();
+                        assertEquals(HttpResponseStatus.OK.code(), response.statusCode());
+                        assertEquals(response.body().getJsonArray("topics").size(), 2);
+                        assertTrue(response.body().getJsonArray("topics").contains(topic));
+                        assertTrue(response.body().getJsonArray("topics").contains(topic2));
+                        assertEquals(response.body().getJsonArray("partitions").size(), 2);
+                        assertEquals(response.body().getJsonArray("partitions").getJsonObject(0).getJsonArray(topic2).size(), 4);
+                        assertEquals(response.body().getJsonArray("partitions").getJsonObject(1).getJsonArray(topic).size(), 1);
+                        listSubscriptions.complete(true);
+                        context.completeNow();
+                    });
+                });
+
+        assertTrue(context.awaitCompletion(TEST_TIMEOUT, TimeUnit.SECONDS));
+    }
 }
\ No newline at end of file
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java b/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java
index 4a4e650..f8446ae 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/OtherServicesTest.java
@@ -73,8 +73,8 @@ public class OtherServicesTest extends HttpBridgeTestBase {
                         JsonObject bridgeResponse = response.body();
 
                         Map<String, Object> paths = bridgeResponse.getJsonObject("paths").getMap();
-                        // subscribe and unsubscribe are using the same endpoint but different methods
-                        int pathsSize = HttpOpenApiOperations.values().length - 1;
+                        // subscribe, list subscribtions and unsubscribe are using the same endpoint but different methods
+                        int pathsSize = HttpOpenApiOperations.values().length - 2;
                         assertEquals(pathsSize, paths.size());
                         assertTrue(paths.containsKey("/consumers/{groupid}"));
                         assertEquals(HttpOpenApiOperations.CREATE_CONSUMER.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}").getJsonObject("post").getString("operationId"));
@@ -87,6 +87,7 @@ public class OtherServicesTest extends HttpBridgeTestBase {
                         assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/subscription"));
                         assertEquals(HttpOpenApiOperations.SUBSCRIBE.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("post").getString("operationId"));
                         assertEquals(HttpOpenApiOperations.UNSUBSCRIBE.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("delete").getString("operationId"));
+                        assertEquals(HttpOpenApiOperations.LIST_SUBSCRIPTIONS.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/subscription").getJsonObject("get").getString("operationId"));
                         assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/assignments"));
                         assertEquals(HttpOpenApiOperations.ASSIGN.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/consumers/{groupid}/instances/{name}/assignments").getJsonObject("post").getString("operationId"));
                         assertTrue(paths.containsKey("/consumers/{groupid}/instances/{name}/records"));
@@ -104,7 +105,7 @@ public class OtherServicesTest extends HttpBridgeTestBase {
                         assertTrue(paths.containsKey("/openapi"));
                         assertEquals(HttpOpenApiOperations.OPENAPI.toString(), bridgeResponse.getJsonObject("paths").getJsonObject("/openapi").getJsonObject("get").getString("operationId"));
                         assertFalse(paths.containsKey("/karel"));
-                        assertEquals(16, bridgeResponse.getJsonObject("definitions").getMap().size());
+                        assertEquals(18, bridgeResponse.getJsonObject("definitions").getMap().size());
                         assertEquals(4, bridgeResponse.getJsonArray("tags").size());
                     });
                     context.completeNow();
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java b/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
index 17f3c20..1130a89 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/services/ConsumerService.java
@@ -71,6 +71,11 @@ public class ConsumerService extends BaseService {
                 .putHeader(CONTENT_TYPE.toString(), BridgeContentType.KAFKA_JSON);
     }
 
+    public HttpRequest<Buffer> listSubscriptionsConsumerRequest(String groupId, String name) {
+        return getRequest(Urls.consumerInstanceSubscription(groupId, name))
+                .putHeader(ACCEPT.toString(), BridgeContentType.KAFKA_JSON);
+    }
+
     public HttpRequest<JsonObject> offsetsRequest(String groupId, String name, JsonObject json) {
         return postRequest(Urls.consumerInstanceOffsets(groupId, name))
                 .putHeader(CONTENT_LENGTH.toString(), String.valueOf(json.toBuffer().length()))