You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by yo...@apache.org on 2023/06/26 01:26:45 UTC
[bookkeeper] branch branch-4.16 updated (9f63cf7e50 -> 8b3f782c59)
This is an automated email from the ASF dual-hosted git repository.
yong pushed a change to branch branch-4.16
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
from 9f63cf7e50 [Branch-4.16] Downgrade grpc and protobuf to avoid introducing breaking change (#4001)
new 14dbfd2d78 Fix data lost when configured multiple ledger directories (#3329)
new 8b3f782c59 Fix trigger GC not work (#3998)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/bookkeeper/bookie/BookieImpl.java | 5 +
.../bookkeeper/bookie/GarbageCollectorThread.java | 5 +-
.../bookie/InterleavedLedgerStorage.java | 2 +-
.../java/org/apache/bookkeeper/bookie/Journal.java | 2 +-
.../apache/bookkeeper/bookie/LedgerStorage.java | 2 +-
.../bookkeeper/bookie/SortedLedgerStorage.java | 2 +-
.../bookie/storage/ldb/DbLedgerStorage.java | 2 +-
.../ldb/SingleDirectoryDbLedgerStorage.java | 8 +-
.../server/http/service/TriggerGCService.java | 74 +++++----
.../bookkeeper/bookie/MockLedgerStorage.java | 2 +-
.../bookie/storage/ldb/DbLedgerStorageTest.java | 183 +++++++++++++++++++++
.../server/http/service/TriggerGCServiceTest.java | 147 +++++++++++++++++
12 files changed, 392 insertions(+), 42 deletions(-)
create mode 100644 bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/service/TriggerGCServiceTest.java
[bookkeeper] 02/02: Fix trigger GC not work (#3998)
Posted by yo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yong pushed a commit to branch branch-4.16
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
commit 8b3f782c59fc4c5a40ffd62ec56538bafae5af5a
Author: Hang Chen <ch...@apache.org>
AuthorDate: Mon Jun 26 09:09:21 2023 +0800
Fix trigger GC not work (#3998)
* Fix trigger gc not work
(cherry picked from commit 147b5bf846189584d6a9b4e20ac9a73e6e617255)
---
.../bookkeeper/bookie/GarbageCollectorThread.java | 5 +-
.../bookie/InterleavedLedgerStorage.java | 2 +-
.../apache/bookkeeper/bookie/LedgerStorage.java | 2 +-
.../bookkeeper/bookie/SortedLedgerStorage.java | 2 +-
.../bookie/storage/ldb/DbLedgerStorage.java | 2 +-
.../ldb/SingleDirectoryDbLedgerStorage.java | 2 +-
.../server/http/service/TriggerGCService.java | 74 ++++++-----
.../bookkeeper/bookie/MockLedgerStorage.java | 2 +-
.../server/http/service/TriggerGCServiceTest.java | 147 +++++++++++++++++++++
9 files changed, 198 insertions(+), 40 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java
index 552592f624..1dc7c4b5a5 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/GarbageCollectorThread.java
@@ -292,12 +292,11 @@ public class GarbageCollectorThread implements Runnable {
}
}
- public void enableForceGC(Boolean forceMajor, Boolean forceMinor) {
+ public void enableForceGC(boolean forceMajor, boolean forceMinor) {
if (forceGarbageCollection.compareAndSet(false, true)) {
LOG.info("Forced garbage collection triggered by thread: {}, forceMajor: {}, forceMinor: {}",
Thread.currentThread().getName(), forceMajor, forceMinor);
- triggerGC(true, forceMajor == null ? suspendMajorCompaction.get() : !forceMajor,
- forceMinor == null ? suspendMinorCompaction.get() : !forceMinor);
+ triggerGC(true, !forceMajor, !forceMinor);
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java
index c6e6010672..2b90e3b008 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/InterleavedLedgerStorage.java
@@ -260,7 +260,7 @@ public class InterleavedLedgerStorage implements CompactableLedgerStorage, Entry
}
@Override
- public void forceGC(Boolean forceMajor, Boolean forceMinor) {
+ public void forceGC(boolean forceMajor, boolean forceMinor) {
gcThread.enableForceGC(forceMajor, forceMinor);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerStorage.java
index 92f0d38c6e..6eca6e0010 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerStorage.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerStorage.java
@@ -232,7 +232,7 @@ public interface LedgerStorage {
/**
* Force trigger Garbage Collection with forceMajor or forceMinor parameter.
*/
- default void forceGC(Boolean forceMajor, Boolean forceMinor) {
+ default void forceGC(boolean forceMajor, boolean forceMinor) {
return;
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java
index 01d14400f0..c3c9e97213 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/SortedLedgerStorage.java
@@ -367,7 +367,7 @@ public class SortedLedgerStorage
}
@Override
- public void forceGC(Boolean forceMajor, Boolean forceMinor) {
+ public void forceGC(boolean forceMajor, boolean forceMinor) {
interleavedLedgerStorage.forceGC(forceMajor, forceMinor);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java
index 5b28332591..8824b1cb6f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorage.java
@@ -506,7 +506,7 @@ public class DbLedgerStorage implements LedgerStorage {
}
@Override
- public void forceGC(Boolean forceMajor, Boolean forceMinor) {
+ public void forceGC(boolean forceMajor, boolean forceMinor) {
ledgerStorageList.stream().forEach(s -> s.forceGC(forceMajor, forceMinor));
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
index c0fff1ec46..f97559dde9 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
@@ -269,7 +269,7 @@ public class SingleDirectoryDbLedgerStorage implements CompactableLedgerStorage
}
@Override
- public void forceGC(Boolean forceMajor, Boolean forceMinor) {
+ public void forceGC(boolean forceMajor, boolean forceMinor) {
gcThread.enableForceGC(forceMajor, forceMinor);
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/TriggerGCService.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/TriggerGCService.java
index 9f01c5c1b4..1f4eea7fb1 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/TriggerGCService.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/server/http/service/TriggerGCService.java
@@ -22,6 +22,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
import java.util.HashMap;
import java.util.Map;
+import org.apache.bookkeeper.bookie.LedgerStorage;
import org.apache.bookkeeper.common.util.JsonUtil;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.http.HttpServer;
@@ -29,6 +30,7 @@ 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.proto.BookieServer;
+import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -61,40 +63,50 @@ public class TriggerGCService implements HttpEndpointService {
@Override
public HttpServiceResponse handle(HttpServiceRequest request) throws Exception {
HttpServiceResponse response = new HttpServiceResponse();
+ try {
+ if (HttpServer.Method.PUT == request.getMethod()) {
+ String requestBody = request.getBody();
+ if (StringUtils.isBlank(requestBody)) {
+ bookieServer.getBookie().getLedgerStorage().forceGC();
+ } else {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> configMap = JsonUtil.fromJson(requestBody, HashMap.class);
+ LedgerStorage ledgerStorage = bookieServer.getBookie().getLedgerStorage();
+ boolean forceMajor = !ledgerStorage.isMajorGcSuspended();
+ boolean forceMinor = !ledgerStorage.isMinorGcSuspended();
- if (HttpServer.Method.PUT == request.getMethod()) {
- String requestBody = request.getBody();
- if (null == requestBody) {
- bookieServer.getBookie().getLedgerStorage().forceGC();
- } else {
- @SuppressWarnings("unchecked")
- Map<String, Object> configMap = JsonUtil.fromJson(requestBody, HashMap.class);
- Boolean forceMajor = (Boolean) configMap.getOrDefault("forceMajor", null);
- Boolean forceMinor = (Boolean) configMap.getOrDefault("forceMinor", null);
- bookieServer.getBookie().getLedgerStorage().forceGC(forceMajor, forceMinor);
- }
+ forceMajor = Boolean.parseBoolean(configMap.getOrDefault("forceMajor", forceMajor).toString());
+ forceMinor = Boolean.parseBoolean(configMap.getOrDefault("forceMinor", forceMinor).toString());
+ ledgerStorage.forceGC(forceMajor, forceMinor);
+ }
- String output = "Triggered GC on BookieServer: " + bookieServer.toString();
- String jsonResponse = JsonUtil.toJson(output);
- if (LOG.isDebugEnabled()) {
- LOG.debug("output body:" + jsonResponse);
- }
- response.setBody(jsonResponse);
- response.setCode(HttpServer.StatusCode.OK);
- return response;
- } else if (HttpServer.Method.GET == request.getMethod()) {
- Boolean isInForceGC = bookieServer.getBookie().getLedgerStorage().isInForceGC();
- Pair<String, String> output = Pair.of("is_in_force_gc", isInForceGC.toString());
- String jsonResponse = JsonUtil.toJson(output);
- if (LOG.isDebugEnabled()) {
- LOG.debug("output body:" + jsonResponse);
+ String output = "Triggered GC on BookieServer: " + bookieServer.getBookieId();
+ String jsonResponse = JsonUtil.toJson(output);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("output body:" + jsonResponse);
+ }
+ response.setBody(jsonResponse);
+ response.setCode(HttpServer.StatusCode.OK);
+ return response;
+ } else if (HttpServer.Method.GET == request.getMethod()) {
+ Boolean isInForceGC = bookieServer.getBookie().getLedgerStorage().isInForceGC();
+ Pair<String, String> output = Pair.of("is_in_force_gc", isInForceGC.toString());
+ String jsonResponse = JsonUtil.toJson(output);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("output body:" + jsonResponse);
+ }
+ response.setBody(jsonResponse);
+ response.setCode(HttpServer.StatusCode.OK);
+ return response;
+ } else {
+ response.setCode(HttpServer.StatusCode.METHOD_NOT_ALLOWED);
+ response.setBody("Not allowed method. Should be PUT to trigger GC, Or GET to get Force GC state.");
+ return response;
}
- response.setBody(jsonResponse);
- response.setCode(HttpServer.StatusCode.OK);
- return response;
- } else {
- response.setCode(HttpServer.StatusCode.NOT_FOUND);
- response.setBody("Not found method. Should be PUT to trigger GC, Or GET to get Force GC state.");
+ } catch (Exception e) {
+ LOG.error("Failed to handle the request, method: {}, body: {} ", request.getMethod(), request.getBody(), e);
+ response.setCode(HttpServer.StatusCode.BAD_REQUEST);
+ response.setBody("Failed to handle the request, exception: " + e.getMessage());
return response;
}
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/MockLedgerStorage.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/MockLedgerStorage.java
index 0b14c395a8..e1979dbdc4 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/MockLedgerStorage.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/MockLedgerStorage.java
@@ -271,7 +271,7 @@ public class MockLedgerStorage implements CompactableLedgerStorage {
}
@Override
- public void forceGC(Boolean forceMajor, Boolean forceMinor) {
+ public void forceGC(boolean forceMajor, boolean forceMinor) {
CompactableLedgerStorage.super.forceGC(forceMajor, forceMinor);
}
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/service/TriggerGCServiceTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/service/TriggerGCServiceTest.java
new file mode 100644
index 0000000000..dca0466e56
--- /dev/null
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/server/http/service/TriggerGCServiceTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.bookie.LedgerStorage;
+import org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage;
+import org.apache.bookkeeper.conf.ServerConfiguration;
+import org.apache.bookkeeper.http.HttpServer;
+import org.apache.bookkeeper.http.service.HttpServiceRequest;
+import org.apache.bookkeeper.http.service.HttpServiceResponse;
+import org.apache.bookkeeper.proto.BookieServer;
+import org.junit.Before;
+import org.junit.Test;
+
+
+/**
+ * Unit test for {@link TriggerGCService}.
+ */
+@Slf4j
+public class TriggerGCServiceTest {
+ private TriggerGCService service;
+ private BookieServer mockBookieServer;
+ private LedgerStorage mockLedgerStorage;
+
+ @Before
+ public void setup() {
+ this.mockBookieServer = mock(BookieServer.class, RETURNS_DEEP_STUBS);
+ this.mockLedgerStorage = mock(DbLedgerStorage.class);
+ when(mockBookieServer.getBookie().getLedgerStorage()).thenReturn(mockLedgerStorage);
+ when(mockLedgerStorage.isInForceGC()).thenReturn(false);
+ when(mockLedgerStorage.isMajorGcSuspended()).thenReturn(false);
+ when(mockLedgerStorage.isMinorGcSuspended()).thenReturn(false);
+ this.service = new TriggerGCService(new ServerConfiguration(), mockBookieServer);
+ }
+
+ @Test
+ public void testHandleRequest() throws Exception {
+
+ // test empty put body
+ HttpServiceRequest request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.PUT);
+ HttpServiceResponse resp = service.handle(request);
+ assertEquals(HttpServer.StatusCode.OK.getValue(), resp.getStatusCode());
+ assertEquals("\"Triggered GC on BookieServer: " + mockBookieServer.getBookieId() + "\"",
+ resp.getBody());
+
+ // test invalid put json body
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.PUT);
+ request.setBody("test");
+ resp = service.handle(request);
+ assertEquals(HttpServer.StatusCode.BAD_REQUEST.getValue(), resp.getStatusCode());
+ assertEquals("Failed to handle the request, exception: Failed to deserialize Object from Json string",
+ resp.getBody());
+
+ // test forceMajor and forceMinor not set
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.PUT);
+ request.setBody("{\"test\":1}");
+ resp = service.handle(request);
+ verify(mockLedgerStorage, times(1)).forceGC(eq(true), eq(true));
+ assertEquals(HttpServer.StatusCode.OK.getValue(), resp.getStatusCode());
+ assertEquals("\"Triggered GC on BookieServer: " + mockBookieServer.getBookieId() + "\"",
+ resp.getBody());
+
+ // test forceMajor set, but forceMinor not set
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.PUT);
+ request.setBody("{\"test\":1,\"forceMajor\":true}");
+ resp = service.handle(request);
+ verify(mockLedgerStorage, times(2)).forceGC(eq(true), eq(true));
+ assertEquals(HttpServer.StatusCode.OK.getValue(), resp.getStatusCode());
+ assertEquals("\"Triggered GC on BookieServer: " + mockBookieServer.getBookieId() + "\"",
+ resp.getBody());
+
+ // test forceMajor set, but forceMinor not set
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.PUT);
+ request.setBody("{\"test\":1,\"forceMajor\":\"true\"}");
+ resp = service.handle(request);
+ verify(mockLedgerStorage, times(3)).forceGC(eq(true), eq(true));
+ assertEquals(HttpServer.StatusCode.OK.getValue(), resp.getStatusCode());
+ assertEquals("\"Triggered GC on BookieServer: " + mockBookieServer.getBookieId() + "\"",
+ resp.getBody());
+
+ // test forceMajor set to false, and forMinor not set
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.PUT);
+ request.setBody("{\"test\":1,\"forceMajor\":false}");
+ resp = service.handle(request);
+ verify(mockLedgerStorage, times(1)).forceGC(eq(false), eq(true));
+ assertEquals(HttpServer.StatusCode.OK.getValue(), resp.getStatusCode());
+ assertEquals("\"Triggered GC on BookieServer: " + mockBookieServer.getBookieId() + "\"",
+ resp.getBody());
+
+ // test forceMajor not set and forMinor set
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.PUT);
+ request.setBody("{\"test\":1,\"forceMinor\":true}");
+ resp = service.handle(request);
+ verify(mockLedgerStorage, times(4)).forceGC(eq(true), eq(true));
+ assertEquals(HttpServer.StatusCode.OK.getValue(), resp.getStatusCode());
+ assertEquals("\"Triggered GC on BookieServer: " + mockBookieServer.getBookieId() + "\"",
+ resp.getBody());
+
+ // test get gc
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.GET);
+ resp = service.handle(request);
+ assertEquals(HttpServer.StatusCode.OK.getValue(), resp.getStatusCode());
+ assertEquals("{\n \"is_in_force_gc\" : \"false\"\n}", resp.getBody());
+
+ // test invalid method type
+ request = new HttpServiceRequest();
+ request.setMethod(HttpServer.Method.POST);
+ resp = service.handle(request);
+ assertEquals(HttpServer.StatusCode.METHOD_NOT_ALLOWED.getValue(), resp.getStatusCode());
+ assertEquals("Not allowed method. Should be PUT to trigger GC, Or GET to get Force GC state.",
+ resp.getBody());
+ }
+
+}
[bookkeeper] 01/02: Fix data lost when configured multiple ledger directories (#3329)
Posted by yo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yong pushed a commit to branch branch-4.16
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
commit 14dbfd2d787e73c1b2c2469841c2b98e1370f2da
Author: Hang Chen <ch...@apache.org>
AuthorDate: Sun Jun 25 23:58:57 2023 +0800
Fix data lost when configured multiple ledger directories (#3329)
(cherry picked from commit 8a76703ee44b1f5af9eaedd68a53368dbf5855f0)
---
.../org/apache/bookkeeper/bookie/BookieImpl.java | 5 +
.../java/org/apache/bookkeeper/bookie/Journal.java | 2 +-
.../ldb/SingleDirectoryDbLedgerStorage.java | 6 +-
.../bookie/storage/ldb/DbLedgerStorageTest.java | 183 +++++++++++++++++++++
4 files changed, 194 insertions(+), 2 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java
index 0628ec28af..e5520185f3 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/BookieImpl.java
@@ -1285,4 +1285,9 @@ public class BookieImpl extends BookieCriticalThread implements Bookie {
}
}
}
+
+ @VisibleForTesting
+ public List<Journal> getJournals() {
+ return this.journals;
+ }
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java
index d8df972878..9a056ca67b 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Journal.java
@@ -226,7 +226,7 @@ public class Journal extends BookieCriticalThread implements CheckpointSource {
* The last mark should first be max journal log id,
* and then max log position in max journal log.
*/
- void readLog() {
+ public void readLog() {
byte[] buff = new byte[16];
ByteBuffer bb = ByteBuffer.wrap(buff);
LogMark mark = new LogMark();
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
index a140db7010..c0fff1ec46 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/SingleDirectoryDbLedgerStorage.java
@@ -142,6 +142,7 @@ public class SingleDirectoryDbLedgerStorage implements CompactableLedgerStorage
private final long maxReadAheadBytesSize;
private final Counter flushExecutorTime;
+ private final boolean singleLedgerDirs;
public SingleDirectoryDbLedgerStorage(ServerConfiguration conf, LedgerManager ledgerManager,
LedgerDirsManager ledgerDirsManager, LedgerDirsManager indexDirsManager,
@@ -172,6 +173,7 @@ public class SingleDirectoryDbLedgerStorage implements CompactableLedgerStorage
this.writeCacheMaxSize = writeCacheSize;
this.writeCache = new WriteCache(allocator, writeCacheMaxSize / 2);
this.writeCacheBeingFlushed = new WriteCache(allocator, writeCacheMaxSize / 2);
+ this.singleLedgerDirs = conf.getLedgerDirs().length == 1;
readCacheMaxSize = readCacheSize;
this.readAheadCacheBatchSize = readAheadCacheBatchSize;
@@ -895,7 +897,9 @@ public class SingleDirectoryDbLedgerStorage implements CompactableLedgerStorage
public void flush() throws IOException {
Checkpoint cp = checkpointSource.newCheckpoint();
checkpoint(cp);
- checkpointSource.checkpointComplete(cp, true);
+ if (singleLedgerDirs) {
+ checkpointSource.checkpointComplete(cp, true);
+ }
}
@Override
diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java
index 2a7e8e2869..65f11e5d6a 100644
--- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java
+++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/ldb/DbLedgerStorageTest.java
@@ -31,16 +31,21 @@ import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.util.ReferenceCountUtil;
import java.io.File;
+import java.io.FileInputStream;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.List;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.Bookie.NoEntryException;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.bookie.BookieImpl;
+import org.apache.bookkeeper.bookie.CheckpointSource;
+import org.apache.bookkeeper.bookie.CheckpointSourceList;
import org.apache.bookkeeper.bookie.DefaultEntryLogger;
import org.apache.bookkeeper.bookie.EntryLocation;
import org.apache.bookkeeper.bookie.LedgerDirsManager;
import org.apache.bookkeeper.bookie.LedgerStorage;
+import org.apache.bookkeeper.bookie.LogMark;
import org.apache.bookkeeper.bookie.TestBookieImpl;
import org.apache.bookkeeper.bookie.storage.EntryLogger;
import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -639,4 +644,182 @@ public class DbLedgerStorageTest {
storage = (DbLedgerStorage) new TestBookieImpl(conf).getLedgerStorage();
}
+
+ @Test
+ public void testMultiLedgerDirectoryCheckpoint() throws Exception {
+ int gcWaitTime = 1000;
+ File firstDir = new File(tmpDir, "dir1");
+ File secondDir = new File(tmpDir, "dir2");
+ ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+ conf.setGcWaitTime(gcWaitTime);
+ conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 4);
+ conf.setProperty(DbLedgerStorage.READ_AHEAD_CACHE_MAX_SIZE_MB, 4);
+ conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ conf.setLedgerDirNames(new String[] { firstDir.getCanonicalPath(), secondDir.getCanonicalPath() });
+
+ BookieImpl bookie = new TestBookieImpl(conf);
+ ByteBuf entry1 = Unpooled.buffer(1024);
+ entry1.writeLong(1); // ledger id
+ entry1.writeLong(2); // entry id
+ entry1.writeBytes("entry-1".getBytes());
+
+ bookie.getLedgerStorage().addEntry(entry1);
+ // write one entry to first ledger directory and flush with logMark(1, 2),
+ // only the first ledger directory should have lastMark
+ bookie.getJournals().get(0).getLastLogMark().getCurMark().setLogMark(1, 2);
+ ((DbLedgerStorage) bookie.getLedgerStorage()).getLedgerStorageList().get(0).flush();
+
+ File firstDirMark = new File(firstDir + "/current", "lastMark");
+ File secondDirMark = new File(secondDir + "/current", "lastMark");
+
+ // LedgerStorage flush won't trigger lastMark update due to two ledger directories configured
+ try {
+ readLogMark(firstDirMark);
+ readLogMark(secondDirMark);
+ fail();
+ } catch (Exception e) {
+ //
+ }
+
+ // write the second entry to second leger directory and flush with log(4, 5),
+ // the fist ledger directory's lastMark is (1, 2) and the second ledger directory's lastMark is (4, 5);
+ ByteBuf entry2 = Unpooled.buffer(1024);
+ entry2.writeLong(2); // ledger id
+ entry2.writeLong(1); // entry id
+ entry2.writeBytes("entry-2".getBytes());
+
+ bookie.getLedgerStorage().addEntry(entry2);
+ // write one entry to first ledger directory and flush with logMark(1, 2),
+ // only the first ledger directory should have lastMark
+ bookie.getJournals().get(0).getLastLogMark().getCurMark().setLogMark(4, 5);
+ ((DbLedgerStorage) bookie.getLedgerStorage()).getLedgerStorageList().get(1).flush();
+
+ // LedgerStorage flush won't trigger lastMark update due to two ledger directories configured
+ try {
+ readLogMark(firstDirMark);
+ readLogMark(secondDirMark);
+ fail();
+ } catch (Exception e) {
+ //
+ }
+
+ // The dbLedgerStorage flush also won't trigger lastMark update due to two ledger directories configured.
+ bookie.getLedgerStorage().flush();
+ try {
+ readLogMark(firstDirMark);
+ readLogMark(secondDirMark);
+ fail();
+ } catch (Exception e) {
+ //
+ }
+
+ // trigger checkpoint simulate SyncThread do checkpoint.
+ CheckpointSource checkpointSource = new CheckpointSourceList(bookie.getJournals());
+ bookie.getJournals().get(0).getLastLogMark().getCurMark().setLogMark(7, 8);
+ CheckpointSource.Checkpoint checkpoint = checkpointSource.newCheckpoint();
+ checkpointSource.checkpointComplete(checkpoint, false);
+
+ try {
+ LogMark firstLogMark = readLogMark(firstDirMark);
+ LogMark secondLogMark = readLogMark(secondDirMark);
+ assertEquals(7, firstLogMark.getLogFileId());
+ assertEquals(8, firstLogMark.getLogFileOffset());
+ assertEquals(7, secondLogMark.getLogFileId());
+ assertEquals(8, secondLogMark.getLogFileOffset());
+ } catch (Exception e) {
+ fail();
+ }
+
+ // test replay journal lastMark, to make sure we get the right LastMark position
+ bookie.getJournals().get(0).getLastLogMark().readLog();
+ LogMark logMark = bookie.getJournals().get(0).getLastLogMark().getCurMark();
+ assertEquals(7, logMark.getLogFileId());
+ assertEquals(8, logMark.getLogFileOffset());
+ }
+
+ private LogMark readLogMark(File file) throws IOException {
+ byte[] buff = new byte[16];
+ ByteBuffer bb = ByteBuffer.wrap(buff);
+ LogMark mark = new LogMark();
+ try (FileInputStream fis = new FileInputStream(file)) {
+ int bytesRead = fis.read(buff);
+ if (bytesRead != 16) {
+ throw new IOException("Couldn't read enough bytes from lastMark."
+ + " Wanted " + 16 + ", got " + bytesRead);
+ }
+ }
+ bb.clear();
+ mark.readLogMark(bb);
+
+ return mark;
+ }
+
+ @Test
+ public void testSingleLedgerDirectoryCheckpoint() throws Exception {
+ int gcWaitTime = 1000;
+ File ledgerDir = new File(tmpDir, "dir");
+ ServerConfiguration conf = TestBKConfiguration.newServerConfiguration();
+ conf.setGcWaitTime(gcWaitTime);
+ conf.setProperty(DbLedgerStorage.WRITE_CACHE_MAX_SIZE_MB, 4);
+ conf.setProperty(DbLedgerStorage.READ_AHEAD_CACHE_MAX_SIZE_MB, 4);
+ conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
+ conf.setLedgerDirNames(new String[] { ledgerDir.getCanonicalPath() });
+
+ BookieImpl bookie = new TestBookieImpl(conf);
+ ByteBuf entry1 = Unpooled.buffer(1024);
+ entry1.writeLong(1); // ledger id
+ entry1.writeLong(2); // entry id
+ entry1.writeBytes("entry-1".getBytes());
+ bookie.getLedgerStorage().addEntry(entry1);
+
+ bookie.getJournals().get(0).getLastLogMark().getCurMark().setLogMark(1, 2);
+ ((DbLedgerStorage) bookie.getLedgerStorage()).getLedgerStorageList().get(0).flush();
+
+ File ledgerDirMark = new File(ledgerDir + "/current", "lastMark");
+ try {
+ LogMark logMark = readLogMark(ledgerDirMark);
+ assertEquals(1, logMark.getLogFileId());
+ assertEquals(2, logMark.getLogFileOffset());
+ } catch (Exception e) {
+ fail();
+ }
+
+ ByteBuf entry2 = Unpooled.buffer(1024);
+ entry2.writeLong(2); // ledger id
+ entry2.writeLong(1); // entry id
+ entry2.writeBytes("entry-2".getBytes());
+
+ bookie.getLedgerStorage().addEntry(entry2);
+ // write one entry to first ledger directory and flush with logMark(1, 2),
+ // only the first ledger directory should have lastMark
+ bookie.getJournals().get(0).getLastLogMark().getCurMark().setLogMark(4, 5);
+
+ bookie.getLedgerStorage().flush();
+ try {
+ LogMark logMark = readLogMark(ledgerDirMark);
+ assertEquals(4, logMark.getLogFileId());
+ assertEquals(5, logMark.getLogFileOffset());
+ } catch (Exception e) {
+ fail();
+ }
+
+ CheckpointSource checkpointSource = new CheckpointSourceList(bookie.getJournals());
+ bookie.getJournals().get(0).getLastLogMark().getCurMark().setLogMark(7, 8);
+ CheckpointSource.Checkpoint checkpoint = checkpointSource.newCheckpoint();
+ checkpointSource.checkpointComplete(checkpoint, false);
+
+ try {
+ LogMark firstLogMark = readLogMark(ledgerDirMark);
+ assertEquals(7, firstLogMark.getLogFileId());
+ assertEquals(8, firstLogMark.getLogFileOffset());
+ } catch (Exception e) {
+ fail();
+ }
+
+ // test replay journal lastMark, to make sure we get the right LastMark position
+ bookie.getJournals().get(0).getLastLogMark().readLog();
+ LogMark logMark = bookie.getJournals().get(0).getLastLogMark().getCurMark();
+ assertEquals(7, logMark.getLogFileId());
+ assertEquals(8, logMark.getLogFileOffset());
+ }
}