Adds the bookie cookie query service for http api

bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpRouter.java

@@ -56,6 +56,7 @@ public abstract class HttpRouter<Handler> {
     public static final String BOOKIE_INFO                  = "/api/v1/bookie/info";
     public static final String CLUSTER_INFO                  = "/api/v1/bookie/cluster_info";
     public static final String ENTRY_LOCATION_COMPACT       = "/api/v1/bookie/entry_location_compact";
+    public static final String BOOKIE_COOKIE                = "/api/v1/bookie/cookie";
     // autorecovery
     public static final String AUTORECOVERY_STATUS          = "/api/v1/autorecovery/status";
     public static final String RECOVERY_BOOKIE              = "/api/v1/autorecovery/bookie";
@@ -100,6 +101,7 @@ public HttpRouter(AbstractHttpHandlerFactory<Handler> handlerFactory) {
             handlerFactory.newHandler(HttpServer.ApiType.RESUME_GC_COMPACTION));
         this.endpointHandlers.put(ENTRY_LOCATION_COMPACT,
                 handlerFactory.newHandler(HttpServer.ApiType.TRIGGER_ENTRY_LOCATION_COMPACT));
+        this.endpointHandlers.put(BOOKIE_COOKIE, handlerFactory.newHandler(HttpServer.ApiType.BOOKIE_COOKIE));
 
         // autorecovery
         this.endpointHandlers.put(AUTORECOVERY_STATUS, handlerFactory

bookkeeper-http/http-server/src/main/java/org/apache/bookkeeper/http/HttpServer.java

@@ -91,6 +91,7 @@ enum ApiType {
         RESUME_GC_COMPACTION,
         SUSPEND_GC_COMPACTION,
         TRIGGER_ENTRY_LOCATION_COMPACT,
+        BOOKIE_COOKIE,
         // autorecovery
         AUTORECOVERY_STATUS,
         RECOVERY_BOOKIE,

bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/BKHttpServiceProvider.java

@@ -38,6 +38,7 @@
 import org.apache.bookkeeper.replication.Auditor;
 import org.apache.bookkeeper.replication.AutoRecoveryMain;
 import org.apache.bookkeeper.server.http.service.AutoRecoveryStatusService;
+import org.apache.bookkeeper.server.http.service.BookieCookieService;
 import org.apache.bookkeeper.server.http.service.BookieInfoService;
 import org.apache.bookkeeper.server.http.service.BookieIsReadyService;
 import org.apache.bookkeeper.server.http.service.BookieSanityService;
@@ -238,6 +239,8 @@ public HttpEndpointService provideHttpEndpointService(ApiType type) {
                 return new ResumeCompactionService(bookieServer);
             case TRIGGER_ENTRY_LOCATION_COMPACT:
                 return new TriggerLocationCompactService(bookieServer);
+            case BOOKIE_COOKIE:
+                return new BookieCookieService(configuration);
 
             // autorecovery
             case AUTORECOVERY_STATUS:

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

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.bookkeeper.server.http.service;
+
+import java.net.UnknownHostException;
+import java.util.Map;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.bookie.Cookie;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.http.service.HttpEndpointService;
+import org.apache.bookkeeper.http.service.HttpServiceRequest;
+import org.apache.bookkeeper.http.service.HttpServiceResponse;
+import org.apache.bookkeeper.meta.MetadataDrivers;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.net.BookieSocketAddress;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BookieCookieService implements HttpEndpointService {
+    static final Logger LOG = LoggerFactory.getLogger(BookieCookieService.class);
+    private final ServerConfiguration conf;
+
+    public BookieCookieService(ServerConfiguration conf) {
+        this.conf = conf;
+    }
+
+    @SuppressWarnings("checkstyle:RegexpSingleline")
+    @Override
+    public HttpServiceResponse handle(HttpServiceRequest request) throws Exception {
+        Map<String, String> params = request.getParams();
+        if (params == null || !params.containsKey("bookie_id")) {
+            return new HttpServiceResponse("Not found bookie id. Should provide bookie_id=<ip:port>",
+                    HttpServer.StatusCode.BAD_REQUEST);
+        }
+        if (request.getMethod() != HttpServer.Method.GET) {
+            return new HttpServiceResponse("Not support for cookie update.", HttpServer.StatusCode.BAD_REQUEST);
+        }
+
+        String bookieIdStr = params.get("bookie_id");
+        try {
+            new BookieSocketAddress(bookieIdStr);
+        } catch (UnknownHostException nhe) {
+            return new HttpServiceResponse("Illegal bookie id. Should provide bookie_id=<ip:port>",
+                    HttpServer.StatusCode.BAD_REQUEST);
+        }
+
+        BookieId bookieId = BookieId.parse(bookieIdStr);
+        return MetadataDrivers.runFunctionWithRegistrationManager(conf, registrationManager -> {
+            try {
+                if (request.getMethod() == HttpServer.Method.GET) {
+                    Versioned<Cookie> cookie = Cookie.readFromRegistrationManager(registrationManager, bookieId);
+                    return new HttpServiceResponse(cookie.getValue().toString(), HttpServer.StatusCode.OK);
+                } else {
+                    return new HttpServiceResponse("Method not allowed. Should be GET method",
+                            HttpServer.StatusCode.METHOD_NOT_ALLOWED);
+                }
+            } catch (BookieException.CookieNotFoundException e) {
+                return new HttpServiceResponse("Not found cookie: " + bookieId, HttpServer.StatusCode.NOT_FOUND);
+            } catch (BookieException e) {
+                LOG.error("Failed to get bookie cookie: ", e);
+                return new HttpServiceResponse("Request failed, e:" + e.getMessage(),
+                        HttpServer.StatusCode.INTERNAL_ERROR);
+            }
+        });
+    }
+}

bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/TestHttpService.java

@@ -1270,4 +1270,44 @@ public void testTriggerEntryLocationCompactService() throws Exception {
         HttpServiceResponse response7 = triggerEntryLocationCompactService.handle(request7);
         assertEquals(HttpServer.StatusCode.METHOD_NOT_ALLOWED.getValue(), response7.getStatusCode());
     }
+
+    @SuppressWarnings("checkstyle:RegexpSingleline")
+    @Test
+    public void testBookieCookieService() throws Exception {
+        runFunctionWithRegistrationManager(baseConf, registrationManager -> {
+            try {
+                String bookieId = getBookie(0).toString();
+                Versioned<Cookie> cookieFromZk = Cookie.readFromRegistrationManager(registrationManager,
+                        BookieId.parse(bookieId));
+                HttpEndpointService bookieCookieService = bkHttpServiceProvider.provideHttpEndpointService(
+                        HttpServer.ApiType.BOOKIE_COOKIE);
+                Map<String, String> params = new HashMap<>();
+                // empty params
+                HttpServiceRequest request = new HttpServiceRequest(null, HttpServer.Method.GET, params);
+                HttpServiceResponse response = bookieCookieService.handle(request);
+                assertEquals(response.getStatusCode(), HttpServer.StatusCode.BAD_REQUEST.getValue());
+                assertEquals(response.getBody(), "Not found bookie id. Should provide bookie_id=<ip:port>");
+                // invalid params
+                params.put("bookie_id", "bookie_id");
+                response = bookieCookieService.handle(request);
+                assertEquals(response.getStatusCode(), HttpServer.StatusCode.BAD_REQUEST.getValue());
+                assertEquals(response.getBody(), "Illegal bookie id. Should provide bookie_id=<ip:port>");
+
+                // cookie not found
+                params.put("bookie_id", "127.2.1.0:3181");
+                response = bookieCookieService.handle(request);
+                assertEquals(response.getStatusCode(), HttpServer.StatusCode.NOT_FOUND.getValue());
+
+                params.put("bookie_id", bookieId);
+                // GET cookie
+                HttpServiceRequest request1 = new HttpServiceRequest(null, HttpServer.Method.GET, params);
+                HttpServiceResponse response1 = bookieCookieService.handle(request1);
+                assertEquals(response1.getStatusCode(), HttpServer.StatusCode.OK.getValue());
+                assertEquals(cookieFromZk.getValue().toString(), response1.getBody());
+                return true;
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        });
+    }
 }