Fixes vertx can not receive the request body

bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/service/NullHttpService.java

@@ -30,6 +30,9 @@ public class NullHttpService implements HttpEndpointService {
 
     @Override
     public HttpServiceResponse handle(HttpServiceRequest request) {
+        if (request.getBody() != null) {
+            return new HttpServiceResponse(request.getBody(), HttpServer.StatusCode.OK);
+        }
         return new HttpServiceResponse(CONTENT, HttpServer.StatusCode.OK);
     }
 }

bookkeeper-http/vertx-http-server/src/main/java/org/apache/bookkeeper/http/vertx/VertxHttpServer.java

@@ -24,6 +24,7 @@
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Vertx;
 import io.vertx.ext.web.Router;
+import io.vertx.ext.web.handler.BodyHandler;
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
@@ -66,6 +67,7 @@ public boolean startServer(int port) {
         CompletableFuture<AsyncResult<io.vertx.core.http.HttpServer>> future = new CompletableFuture<>();
         VertxHttpHandlerFactory handlerFactory = new VertxHttpHandlerFactory(httpServiceProvider);
         Router router = Router.router(vertx);
+        router.route().handler(BodyHandler.create());
         HttpRouter<VertxAbstractHandler> requestRouter = new HttpRouter<VertxAbstractHandler>(handlerFactory) {
             @Override
             public void bindHandler(String endpoint, VertxAbstractHandler handler) {

bookkeeper-http/vertx-http-server/src/test/java/org/apache/bookkeeper/http/vertx/TestVertxHttpServer.java

@@ -26,8 +26,10 @@
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
+import java.io.OutputStream;
 import java.net.HttpURLConnection;
 import java.net.URL;
+import java.nio.charset.StandardCharsets;
 
 import org.apache.bookkeeper.http.HttpRouter;
 import org.apache.bookkeeper.http.HttpServer;
@@ -81,12 +83,37 @@ public void testHttpMethods() throws Exception {
         httpServer.stopServer();
     }
 
-    // HTTP request
+    @Test
+    public void testHttpMethodsWithBody() throws IOException {
+        VertxHttpServer httpServer = new VertxHttpServer();
+        HttpServiceProvider httpServiceProvider = NullHttpServiceProvider.getInstance();
+        httpServer.initialize(httpServiceProvider);
+        assertTrue(httpServer.startServer(0));
+        int port = httpServer.getListeningPort();
+        String body = "{\"bookie_src\": \"localhost:3181\"}";
+        HttpResponse httpResponse = send(getUrl(port, HttpRouter.DECOMMISSION), HttpServer.Method.PUT, body);
+        assertEquals(HttpServer.StatusCode.OK.getValue(), httpResponse.responseCode);
+        assertEquals(body, httpResponse.responseBody);
+        httpServer.stopServer();
+    }
+
     private HttpResponse send(String url, HttpServer.Method method) throws IOException {
+        return send(url, method, "");
+    }
+
+    // HTTP request
+    private HttpResponse send(String url, HttpServer.Method method, String body) throws IOException {
         URL obj = new URL(url);
         HttpURLConnection con = (HttpURLConnection) obj.openConnection();
         // optional, default is GET
         con.setRequestMethod(method.toString());
+        if (body != "") {
+            con.setDoOutput(true);
+            con.setFixedLengthStreamingMode(body.length());
+            OutputStream outputStream = con.getOutputStream();
+            outputStream.write(body.getBytes(StandardCharsets.UTF_8));
+            outputStream.flush();
+        }
         int responseCode = con.getResponseCode();
         StringBuilder response = new StringBuilder();
         BufferedReader in = null;

stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterInitializer.java

@@ -20,6 +20,7 @@
 
 import static org.apache.bookkeeper.stream.storage.StorageConstants.ZK_METADATA_ROOT_PATH;
 import static org.apache.bookkeeper.stream.storage.StorageConstants.getSegmentsRootPath;
+
 import com.google.common.base.Strings;
 import java.net.URI;
 import java.util.Optional;