Delete cookie as part of decommission API (#4592)

Implement the logic in [BP-68:](#4500) Delete cookie as part of decommission API.

Author: SongOf

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

@@ -19,9 +19,12 @@
 package org.apache.bookkeeper.server.http.service;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithRegistrationManager;
 
 import java.util.HashMap;
 import java.util.concurrent.ExecutorService;
+import org.apache.bookkeeper.bookie.BookieException;
+import org.apache.bookkeeper.bookie.Cookie;
 import org.apache.bookkeeper.client.BookKeeperAdmin;
 import org.apache.bookkeeper.common.util.JsonUtil;
 import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -30,6 +33,7 @@
 import org.apache.bookkeeper.http.service.HttpServiceRequest;
 import org.apache.bookkeeper.http.service.HttpServiceResponse;
 import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.versioning.Versioned;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -74,12 +78,30 @@ public HttpServiceResponse handle(HttpServiceRequest request) throws Exception {
             if (configMap != null && configMap.containsKey("bookie_src")) {
                 try {
                     BookieId bookieSrc = BookieId.parse(configMap.get("bookie_src"));
+                    boolean deleteCookie = Boolean.parseBoolean(configMap.getOrDefault("delete_cookie", "false"));
 
                     executor.execute(() -> {
                         try {
                             LOG.info("Start decommissioning bookie.");
                             bka.decommissionBookie(bookieSrc);
                             LOG.info("Complete decommissioning bookie.");
+                            if (deleteCookie) {
+                                runFunctionWithRegistrationManager(conf, rm -> {
+                                    try {
+                                        Versioned<Cookie> cookie = Cookie.readFromRegistrationManager(rm, bookieSrc);
+                                        cookie.getValue().deleteFromRegistrationManager(rm, bookieSrc,
+                                                cookie.getVersion());
+                                        LOG.info("Cookie of the decommissioned bookie: {} is deleted successfully",
+                                                bookieSrc);
+                                    } catch (BookieException.CookieNotFoundException nne) {
+                                        LOG.warn("No cookie to remove for the decommissioning bookie: {}, "
+                                                + "it could be deleted already", bookieSrc, nne);
+                                    } catch (BookieException be) {
+                                        LOG.error("Error deleting cookie: {}.", bookieSrc, be);
+                                    }
+                                    return true;
+                                });
+                            }
                         } catch (Exception e) {
                             LOG.error("Error handling decommissionBookie: {}.", bookieSrc, e);
                         }

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

@@ -19,6 +19,7 @@
 package org.apache.bookkeeper.server.http;
 
 import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithLedgerManagerFactory;
+import static org.apache.bookkeeper.meta.MetadataDrivers.runFunctionWithRegistrationManager;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -37,8 +38,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import lombok.Cleanup;
+import org.apache.bookkeeper.bookie.BookieException;
 import org.apache.bookkeeper.bookie.BookieResources;
+import org.apache.bookkeeper.bookie.Cookie;
 import org.apache.bookkeeper.bookie.LedgerStorage;
 import org.apache.bookkeeper.client.BookKeeper;
 import org.apache.bookkeeper.client.ClientUtil;
@@ -55,6 +59,7 @@
 import org.apache.bookkeeper.meta.LedgerManagerFactory;
 import org.apache.bookkeeper.meta.LedgerUnderreplicationManager;
 import org.apache.bookkeeper.meta.MetadataBookieDriver;
+import org.apache.bookkeeper.net.BookieId;
 import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.proto.BookieServer;
 import org.apache.bookkeeper.replication.AuditorElector;
@@ -66,6 +71,8 @@
 import org.apache.bookkeeper.server.http.service.ClusterInfoService;
 import org.apache.bookkeeper.stats.NullStatsLogger;
 import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.versioning.Versioned;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -765,15 +772,61 @@ public void testDecommissionService() throws Exception {
         assertEquals(HttpServer.StatusCode.NOT_FOUND.getValue(), response2.getStatusCode());
 
         //3, PUT, with body, should success.
-        String putBody3 = "{\"bookie_src\": \"" + getBookie(1).toString() + "\"}";
+        BookieId bookieId = getBookie(1);
+        String putBody3 = "{\"bookie_src\": \"" + bookieId.toString() + "\"}";
         HttpServiceRequest request3 = new HttpServiceRequest(putBody3, HttpServer.Method.PUT, null);
         // after bookie kill, request should success
         killBookie(1);
         HttpServiceResponse response3 = decommissionService.handle(request3);
         assertEquals(HttpServer.StatusCode.OK.getValue(), response3.getStatusCode());
+        // wait decommission finish
+        Awaitility.await().pollDelay(60, TimeUnit.SECONDS).timeout(70, TimeUnit.SECONDS).untilAsserted(() -> {
+            runFunctionWithRegistrationManager(baseConf, registrationManager -> {
+                Versioned<Cookie> cookieFromZk = null;
+                try {
+                    cookieFromZk = Cookie.readFromRegistrationManager(registrationManager,
+                            bookieId);
+                } catch (BookieException e) {
+                } finally {
+                    assertTrue(cookieFromZk != null);
+                }
+                return true;
+            });
+        });
         stopAuditorElector();
     }
 
+    @Test
+    public void testDecommissionServiceWithDeleteCookie() throws Exception {
+        baseConf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());
+        startReplicationService();
+
+        HttpEndpointService decommissionService = bkHttpServiceProvider
+                .provideHttpEndpointService(HttpServer.ApiType.DECOMMISSION);
+        BookieId bookieId = getBookie(1);
+        String putBody1 = "{\"bookie_src\": \"" + bookieId.toString() + "\", \"delete_cookie\": \"true\"}";
+        HttpServiceRequest request1 = new HttpServiceRequest(putBody1, HttpServer.Method.PUT, null);
+        // after bookie kill, request should success
+        killBookie(1);
+        HttpServiceResponse response1 = decommissionService.handle(request1);
+        assertEquals(HttpServer.StatusCode.OK.getValue(), response1.getStatusCode());
+        // wait decommission finish
+        Awaitility.await().pollInterval(10, TimeUnit.SECONDS).timeout(60, TimeUnit.SECONDS).untilAsserted(() -> {
+            runFunctionWithRegistrationManager(baseConf, registrationManager -> {
+                Versioned<Cookie> cookieFromZk = null;
+                try {
+                    cookieFromZk = Cookie.readFromRegistrationManager(registrationManager,
+                            bookieId);
+                } catch (Exception e) {
+                } finally {
+                    assertTrue(cookieFromZk == null);
+                }
+                return true;
+            });
+        });
+        stopReplicationService();
+    }
+
     @Test
     public void testTriggerGCService() throws Exception {
         baseConf.setMetadataServiceUri(zkUtil.getMetadataServiceUri());