This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 12d4bbdfed Migrate DataFormats from protobuf-java to LightProto (#4779)
12d4bbdfed is described below
commit 12d4bbdfedfa8372638b20268c3b7e0a530ce217
Author: Matteo Merli <[email protected]>
AuthorDate: Thu May 7 21:31:55 2026 -0700
Migrate DataFormats from protobuf-java to LightProto (#4779)
* Fix LedgerHandle.batchReadUnconfirmedAsync: use slog log instead of LOG
The batchReadUnconfirmedAsync method added in #4739 calls LOG.error(...),
but LedgerHandle was migrated to slog and only has a lowercase `log`
field. Master fails to compile.
Convert the call to the slog builder style used elsewhere in the file.
* Migrate DataFormats and DbLedgerStorageDataFormats from protobuf-java to
LightProto
Replace Google's protobuf-java with StreamNative LightProto for the storage
and
metadata formats in `bookkeeper-proto`. The wire protocol
(`BookkeeperProtocol`)
remains on protobuf-java for now.
LightProto generates mutable, reusable, ByteBuf-aware classes with built-in
proto2 TextFormat (de)serialization (via `generateTextFormat=true`), so the
existing TextFormat-based znode payloads (cookies, auditor votes, lock data,
underreplication entries, layout) round-trip byte-identically.
Notable behavior changes:
- `BookKeeper.DigestType.toProtoDigestType` now returns the
LightProto-generated
enum (same constants, different package).
- v3 ledger metadata uses a hand-rolled length-prefixed delimited
writer/reader
matching protobuf's `writeDelimitedTo`/`mergeDelimitedFrom`.
* Add SpotBugs excludes for lightproto-generated classes
The existing exclude `~org.apache.bookkeeper.proto.DataFormats.*` matched
protobuf's nested `DataFormats$LedgerMetadataFormat` etc. LightProto
generates
the same messages as flat top-level classes (`LedgerMetadataFormat` directly
in `org.apache.bookkeeper.proto`), so those weren't excluded and triggered
12 bugs in the generated code (bit-twiddling, exposed internal byte arrays,
etc.) that aren't actionable.
Replace the obsolete `DataFormats.*` exclude with explicit per-message
patterns covering both packages and `LightProtoCodec` (also generated
per-package).
* Fix checkstyle: remove redundant same-package import in MockBookies
---
bookkeeper-proto/pom.xml | 32 ++-
.../java/org/apache/bookkeeper/bookie/Cookie.java | 29 +--
.../bookie/storage/ldb/LedgerMetadataIndex.java | 31 ++-
.../bookie/storage/ldb/LedgersIndexCheckOp.java | 10 +-
.../bookie/storage/ldb/LedgersIndexRebuildOp.java | 10 +-
.../ldb/SingleDirectoryDbLedgerStorage.java | 19 +-
.../org/apache/bookkeeper/client/BookKeeper.java | 12 +-
.../org/apache/bookkeeper/client/LedgerHandle.java | 5 +-
.../bookkeeper/discover/ZKRegistrationClient.java | 36 +--
.../bookkeeper/discover/ZKRegistrationManager.java | 41 +---
.../bookkeeper/meta/LedgerMetadataSerDe.java | 260 +++++++++++----------
.../bookkeeper/meta/ZkLedgerAuditorManager.java | 16 +-
.../meta/ZkLedgerUnderreplicationManager.java | 108 +++++----
.../bookkeeper/proto/checksum/DigestManager.java | 2 +-
.../bookie/BookieInitializationTest.java | 7 +-
.../org/apache/bookkeeper/client/ClientUtil.java | 2 +-
.../bookkeeper/client/ReadLastConfirmedOpTest.java | 2 +-
.../org/apache/bookkeeper/proto/MockBookies.java | 2 +-
.../TestLedgerUnderreplicationManager.java | 17 +-
.../apache/bookkeeper/test/BookieClientTest.java | 14 +-
.../main/resources/bookkeeper/findbugsExclude.xml | 10 +-
.../proto/checksum/DigestManagerBenchmark.java | 2 +-
.../proto/checksum/DigestTypeBenchmark.java | 2 +-
pom.xml | 1 +
24 files changed, 339 insertions(+), 331 deletions(-)
diff --git a/bookkeeper-proto/pom.xml b/bookkeeper-proto/pom.xml
index 21e28cafce..d997fbde43 100644
--- a/bookkeeper-proto/pom.xml
+++ b/bookkeeper-proto/pom.xml
@@ -29,6 +29,10 @@
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-buffer</artifactId>
+ </dependency>
</dependencies>
<build>
<plugins>
@@ -37,10 +41,9 @@
<artifactId>apache-rat-plugin</artifactId>
<configuration>
<excludes>
- <!-- exclude generated file //-->
- <exclude>**/DataFormats.java</exclude>
+ <!-- exclude generated files //-->
<exclude>**/BookkeeperProtocol.java</exclude>
- <exclude>**/DbLedgerStorageDataFormats.java</exclude>
+ <exclude>target/generated-sources/lightproto/**</exclude>
<exclude>**/.checkstyle</exclude>
</excludes>
</configuration>
@@ -52,6 +55,9 @@
<configuration>
<protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
<checkStaleness>true</checkStaleness>
+ <includes>
+ <include>BookkeeperProtocol.proto</include>
+ </includes>
</configuration>
<executions>
<execution>
@@ -61,6 +67,26 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>io.streamnative.lightproto</groupId>
+ <artifactId>lightproto-maven-plugin</artifactId>
+ <version>${lightproto-maven-plugin.version}</version>
+ <configuration>
+ <sources>
+
<source>${project.basedir}/src/main/proto/DataFormats.proto</source>
+
<source>${project.basedir}/src/main/proto/DbLedgerStorageDataFormats.proto</source>
+ </sources>
+
<targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
+ <generateTextFormat>true</generateTextFormat>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java
index b6a8795a6a..d85e672cea 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java
@@ -24,7 +24,6 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.common.base.Joiner;
import com.google.common.collect.Sets;
-import com.google.protobuf.TextFormat;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.EOFException;
@@ -36,6 +35,7 @@ import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.StringReader;
+import java.io.StringWriter;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.Objects;
@@ -46,7 +46,7 @@ import
org.apache.bookkeeper.bookie.BookieException.UnknownBookieIdException;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.discover.RegistrationManager;
import org.apache.bookkeeper.net.BookieId;
-import org.apache.bookkeeper.proto.DataFormats.CookieFormat;
+import org.apache.bookkeeper.proto.CookieFormat;
import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
@@ -181,20 +181,20 @@ public class Cookie {
if (layoutVersion <= 3) {
return toStringVersion3();
}
- CookieFormat.Builder builder = CookieFormat.newBuilder();
- builder.setBookieHost(bookieId);
- builder.setJournalDir(journalDirs);
- builder.setLedgerDirs(ledgerDirs);
+ CookieFormat fmt = new CookieFormat();
+ fmt.setBookieHost(bookieId);
+ fmt.setJournalDir(journalDirs);
+ fmt.setLedgerDirs(ledgerDirs);
if (null != instanceId) {
- builder.setInstanceId(instanceId);
+ fmt.setInstanceId(instanceId);
}
if (null != indexDirs) {
- builder.setIndexDirs(indexDirs);
+ fmt.setIndexDirs(indexDirs);
}
StringBuilder b = new StringBuilder();
b.append(CURRENT_COOKIE_LAYOUT_VERSION).append("\n");
- b.append(builder.build());
+ fmt.writeTextFormatTo(b);
return b.toString();
}
@@ -226,17 +226,18 @@ public class Cookie {
cBuilder.setJournalDirs(reader.readLine());
cBuilder.setLedgerDirs(reader.readLine());
} else if (layoutVersion >= 4) {
- CookieFormat.Builder cfBuilder = CookieFormat.newBuilder();
- TextFormat.merge(reader, cfBuilder);
- CookieFormat data = cfBuilder.build();
+ CookieFormat data = new CookieFormat();
+ StringWriter rest = new StringWriter();
+ reader.transferTo(rest);
+ data.parseFromTextFormat(rest.toString());
cBuilder.setBookieId(data.getBookieHost());
cBuilder.setJournalDirs(data.getJournalDir());
cBuilder.setLedgerDirs(data.getLedgerDirs());
// Since InstanceId is optional
- if (null != data.getInstanceId() &&
!data.getInstanceId().isEmpty()) {
+ if (data.hasInstanceId() && !data.getInstanceId().isEmpty()) {
cBuilder.setInstanceId(data.getInstanceId());
}
- if (null != data.getIndexDirs() && !data.getIndexDirs().isEmpty())
{
+ if (data.hasIndexDirs() && !data.getIndexDirs().isEmpty()) {
cBuilder.setIndexDirs(data.getIndexDirs());
}
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java
index 7162c0c6c2..9037d4d075 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgerMetadataIndex.java
@@ -23,8 +23,8 @@ package org.apache.bookkeeper.bookie.storage.ldb;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
-import com.google.protobuf.ByteString;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
import java.io.Closeable;
import java.io.IOException;
import java.util.AbstractMap.SimpleEntry;
@@ -37,7 +37,6 @@ import java.util.concurrent.locks.ReentrantLock;
import lombok.CustomLog;
import org.apache.bookkeeper.bookie.Bookie;
import org.apache.bookkeeper.bookie.BookieException;
-import
org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorageDataFormats.LedgerData;
import
org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorage.CloseableIterator;
import
org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorageFactory.DbConfigType;
import org.apache.bookkeeper.conf.ServerConfiguration;
@@ -82,7 +81,8 @@ public class LedgerMetadataIndex implements Closeable {
Entry<byte[], byte[]> entry = iterator.next();
long ledgerId = ArrayUtil.getLong(entry.getKey(), 0);
if (ledgerId >= 0) {
- LedgerData ledgerData =
LedgerData.parseFrom(entry.getValue());
+ LedgerData ledgerData = new LedgerData();
+ ledgerData.parseFrom(entry.getValue());
ledgers.put(ledgerId, ledgerData);
ledgersCount.incrementAndGet();
}
@@ -119,7 +119,7 @@ public class LedgerMetadataIndex implements Closeable {
}
public void set(long ledgerId, LedgerData ledgerData) throws IOException {
- ledgerData = LedgerData.newBuilder(ledgerData).setExists(true).build();
+ ledgerData = new LedgerData().copyFrom(ledgerData).setExists(true);
ReentrantLock lock = lockForLedger(ledgerId);
lock.lock();
@@ -170,11 +170,11 @@ public class LedgerMetadataIndex implements Closeable {
lock.lock();
try {
LedgerData ledgerData = get(ledgerId);
- if (ledgerData.getFenced()) {
+ if (ledgerData.isFenced()) {
return false;
}
- LedgerData newLedgerData =
LedgerData.newBuilder(ledgerData).setFenced(true).build();
+ LedgerData newLedgerData = new
LedgerData().copyFrom(ledgerData).setFenced(true);
if (ledgers.put(ledgerId, newLedgerData) == null) {
// Ledger had been deleted
@@ -197,11 +197,11 @@ public class LedgerMetadataIndex implements Closeable {
lock.lock();
try {
LedgerData ledgerData = get(ledgerId);
- if (ledgerData.getLimbo()) {
+ if (ledgerData.isLimbo()) {
return false;
}
- LedgerData newLedgerData =
LedgerData.newBuilder(ledgerData).setLimbo(true).build();
+ LedgerData newLedgerData = new
LedgerData().copyFrom(ledgerData).setLimbo(true);
if (ledgers.put(ledgerId, newLedgerData) == null) {
// Ledger had been deleted
@@ -227,8 +227,8 @@ public class LedgerMetadataIndex implements Closeable {
if (ledgerData == null) {
throw new Bookie.NoLedgerException(ledgerId);
}
- final boolean oldValue = ledgerData.getLimbo();
- LedgerData newLedgerData =
LedgerData.newBuilder(ledgerData).setLimbo(false).build();
+ final boolean oldValue = ledgerData.isLimbo();
+ LedgerData newLedgerData = new
LedgerData().copyFrom(ledgerData).setLimbo(false);
if (ledgers.put(ledgerId, newLedgerData) == null) {
// Ledger had been deleted
@@ -254,14 +254,13 @@ public class LedgerMetadataIndex implements Closeable {
LedgerData ledgerData = ledgers.get(ledgerId);
if (ledgerData == null) {
// New ledger inserted
- ledgerData =
LedgerData.newBuilder().setExists(true).setFenced(false)
- .setMasterKey(ByteString.copyFrom(masterKey)).build();
+ ledgerData = new
LedgerData().setExists(true).setFenced(false).setMasterKey(masterKey);
log.debug().attr("ledgerId", ledgerId).log("Inserting new
ledger");
} else {
- byte[] storedMasterKey =
ledgerData.getMasterKey().toByteArray();
+ byte[] storedMasterKey = ledgerData.getMasterKey();
if (ArrayUtil.isArrayAllZeros(storedMasterKey)) {
// update master key of the ledger
- ledgerData =
LedgerData.newBuilder(ledgerData).setMasterKey(ByteString.copyFrom(masterKey)).build();
+ ledgerData = new
LedgerData().copyFrom(ledgerData).setMasterKey(masterKey);
log.debug()
.attr("storedMasterKey", storedMasterKey)
.attr("newMasterKey", masterKey)
@@ -388,8 +387,8 @@ public class LedgerMetadataIndex implements Closeable {
void setExplicitLac(long ledgerId, ByteBuf lac) throws IOException {
LedgerData ledgerData = ledgers.get(ledgerId);
if (ledgerData != null) {
- LedgerData newLedgerData = LedgerData.newBuilder(ledgerData)
-
.setExplicitLac(ByteString.copyFrom(lac.nioBuffer())).build();
+ LedgerData newLedgerData = new LedgerData().copyFrom(ledgerData)
+ .setExplicitLac(ByteBufUtil.getBytes(lac));
if (ledgers.put(ledgerId, newLedgerData) == null) {
// Ledger had been deleted
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexCheckOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexCheckOp.java
index 8a7df8f565..c9cae7ddff 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexCheckOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexCheckOp.java
@@ -77,17 +77,17 @@ public class LedgersIndexCheckOp {
ctr++;
Map.Entry<byte[], byte[]> entry = iterator.next();
long ledgerId = ArrayUtil.getLong(entry.getKey(), 0);
- DbLedgerStorageDataFormats.LedgerData ledgerData =
-
DbLedgerStorageDataFormats.LedgerData.parseFrom(entry.getValue());
+ LedgerData ledgerData = new LedgerData();
+ ledgerData.parseFrom(entry.getValue());
if (verbose) {
log.info()
.attr("scanned", ctr)
.attr("ledgerId", ledgerId)
- .attr("exists", (ledgerData.hasExists() ?
ledgerData.getExists() : "-"))
- .attr("isFenced", (ledgerData.hasFenced()
? ledgerData.getFenced() : "-"))
+ .attr("exists", (ledgerData.hasExists() ?
ledgerData.isExists() : "-"))
+ .attr("isFenced", (ledgerData.hasFenced()
? ledgerData.isFenced() : "-"))
.attr("masterKey",
(ledgerData.hasMasterKey()
? Base64.getEncoder()
-
.encodeToString(ledgerData.getMasterKey().toByteArray())
+
.encodeToString(ledgerData.getMasterKey())
: "-"))
.attr("explicitLAC",
(ledgerData.hasExplicitLac()
? ledgerData.getExplicitLac() :
"-"))
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexRebuildOp.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexRebuildOp.java
index fc9f94559a..960d9cd8c1 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexRebuildOp.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/ldb/LedgersIndexRebuildOp.java
@@ -21,7 +21,6 @@
package org.apache.bookkeeper.bookie.storage.ldb;
import com.google.common.collect.Lists;
-import com.google.protobuf.ByteString;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
@@ -109,11 +108,10 @@ public class LedgersIndexRebuildOp {
log.info().attr("tempPath", indexTempPath).log("Created
ledgers index at temp location");
for (Long ledgerId : ledgers) {
- DbLedgerStorageDataFormats.LedgerData ledgerData =
-
DbLedgerStorageDataFormats.LedgerData.newBuilder()
- .setExists(true)
- .setFenced(true)
-
.setMasterKey(ByteString.EMPTY).build();
+ LedgerData ledgerData = new LedgerData()
+ .setExists(true)
+ .setFenced(true)
+ .setMasterKey(new byte[0]);
byte[] ledgerArray = new byte[16];
ArrayUtil.setLong(ledgerArray, 0, ledgerId);
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 708b63da62..6778de2f82 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
@@ -26,7 +26,6 @@ import static com.google.common.base.Preconditions.checkState;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
-import com.google.protobuf.ByteString;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
@@ -68,7 +67,6 @@ import
org.apache.bookkeeper.bookie.LedgerDirsManager.LedgerDirsListener;
import org.apache.bookkeeper.bookie.LedgerEntryPage;
import org.apache.bookkeeper.bookie.StateManager;
import org.apache.bookkeeper.bookie.storage.EntryLogger;
-import
org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorageDataFormats.LedgerData;
import org.apache.bookkeeper.bookie.storage.ldb.KeyValueStorage.Batch;
import org.apache.bookkeeper.common.util.MathUtils;
import org.apache.bookkeeper.common.util.Watcher;
@@ -372,9 +370,9 @@ public class SingleDirectoryDbLedgerStorage implements
CompactableLedgerStorage
LedgerData ledgerData = ledgerIndex.get(ledgerId);
log.debug()
.attr("ledgerId", ledgerId)
- .attr("exists", () -> ledgerData.getExists())
+ .attr("exists", () -> ledgerData.isExists())
.log("Ledger exists");
- return ledgerData.getExists();
+ return ledgerData.isExists();
} catch (Bookie.NoLedgerException nle) {
// ledger does not exist
return false;
@@ -426,7 +424,7 @@ public class SingleDirectoryDbLedgerStorage implements
CompactableLedgerStorage
@Override
public boolean isFenced(long ledgerId) throws IOException, BookieException
{
- boolean isFenced = ledgerIndex.get(ledgerId).getFenced();
+ boolean isFenced = ledgerIndex.get(ledgerId).isFenced();
log.debug()
.attr("ledgerId", ledgerId)
@@ -464,7 +462,7 @@ public class SingleDirectoryDbLedgerStorage implements
CompactableLedgerStorage
@Override
public byte[] readMasterKey(long ledgerId) throws IOException,
BookieException {
log.debug().attr("ledgerId", ledgerId).log("Read master key");
- return ledgerIndex.get(ledgerId).getMasterKey().toByteArray();
+ return ledgerIndex.get(ledgerId).getMasterKey();
}
@Override
@@ -1037,8 +1035,7 @@ public class SingleDirectoryDbLedgerStorage implements
CompactableLedgerStorage
return null;
}
log.debug().attr("ledgerId", ledgerId).log("getExplicitLac returned
from LedgerData");
- ByteString persistedLac = ledgerData.getExplicitLac();
-
ledgerInfo.setExplicitLac(Unpooled.wrappedBuffer(persistedLac.toByteArray()));
+
ledgerInfo.setExplicitLac(Unpooled.wrappedBuffer(ledgerData.getExplicitLac()));
return ledgerInfo.getExplicitLac();
}
@@ -1074,8 +1071,8 @@ public class SingleDirectoryDbLedgerStorage implements
CompactableLedgerStorage
@SuppressFBWarnings("RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE")
public long addLedgerToIndex(long ledgerId, boolean isFenced, byte[]
masterKey,
LedgerCache.PageEntriesIterable pages) throws Exception {
- LedgerData ledgerData =
LedgerData.newBuilder().setExists(true).setFenced(isFenced)
- .setMasterKey(ByteString.copyFrom(masterKey)).build();
+ LedgerData ledgerData = new
LedgerData().setExists(true).setFenced(isFenced)
+ .setMasterKey(masterKey);
ledgerIndex.set(ledgerId, ledgerData);
MutableLong numberOfEntries = new MutableLong();
@@ -1217,7 +1214,7 @@ public class SingleDirectoryDbLedgerStorage implements
CompactableLedgerStorage
@Override
public boolean hasLimboState(long ledgerId) throws IOException {
log.debug().attr("ledgerId", ledgerId).log("hasLimboState");
- return ledgerIndex.get(ledgerId).getLimbo();
+ return ledgerIndex.get(ledgerId).isLimbo();
}
@Override
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
index e34a2ebefa..db34b482f2 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
@@ -85,7 +85,7 @@ import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.proto.BookieAddressResolver;
import org.apache.bookkeeper.proto.BookieClient;
import org.apache.bookkeeper.proto.BookieClientImpl;
-import org.apache.bookkeeper.proto.DataFormats;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.EventLoopUtil;
@@ -738,16 +738,16 @@ public class BookKeeper implements
org.apache.bookkeeper.client.api.BookKeeper {
throw new IllegalArgumentException("Unable to convert
digest type " + digestType);
}
}
- public static DataFormats.LedgerMetadataFormat.DigestType
toProtoDigestType(DigestType digestType) {
+ public static LedgerMetadataFormat.DigestType
toProtoDigestType(DigestType digestType) {
switch (digestType) {
case MAC:
- return DataFormats.LedgerMetadataFormat.DigestType.HMAC;
+ return LedgerMetadataFormat.DigestType.HMAC;
case CRC32:
- return DataFormats.LedgerMetadataFormat.DigestType.CRC32;
+ return LedgerMetadataFormat.DigestType.CRC32;
case CRC32C:
- return DataFormats.LedgerMetadataFormat.DigestType.CRC32C;
+ return LedgerMetadataFormat.DigestType.CRC32C;
case DUMMY:
- return DataFormats.LedgerMetadataFormat.DigestType.DUMMY;
+ return LedgerMetadataFormat.DigestType.DUMMY;
default:
throw new IllegalArgumentException("Unable to convert
digest type " + digestType);
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
index e9e63282ff..0a1005f6ce 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java
@@ -982,7 +982,10 @@ public class LedgerHandle implements WriteHandle {
public CompletableFuture<LedgerEntries> batchReadUnconfirmedAsync(long
startEntry, int maxCount, long maxSize) {
// Little sanity check
if (startEntry < 0 || maxCount < 0 || maxSize < 0) {
- LOG.error("IncorrectParameterException on ledgerId:{}
firstEntry:{} when batch read", ledgerId, startEntry);
+ log.error()
+ .attr("ledgerId", ledgerId)
+ .attr("startEntry", startEntry)
+ .log("IncorrectParameterException when batch read");
return FutureUtils.exception(new BKIncorrectParameterException());
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
index f0ff9121c3..8622f667bd 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationClient.java
@@ -26,8 +26,10 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -36,7 +38,6 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
-import java.util.stream.Collectors;
import lombok.AccessLevel;
import lombok.CustomLog;
import lombok.Getter;
@@ -44,7 +45,7 @@ import org.apache.bookkeeper.client.BKException;
import org.apache.bookkeeper.client.BKException.ZKException;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.net.BookieId;
-import org.apache.bookkeeper.proto.DataFormats.BookieServiceInfoFormat;
+import org.apache.bookkeeper.proto.BookieServiceInfoFormat;
import org.apache.bookkeeper.versioning.LongVersion;
import org.apache.bookkeeper.versioning.Version;
import org.apache.bookkeeper.versioning.Version.Occurred;
@@ -322,23 +323,26 @@ public class ZKRegistrationClient implements
RegistrationClient {
return
BookieServiceInfoUtils.buildLegacyBookieServiceInfo(bookieId.toString());
}
- BookieServiceInfoFormat builder =
BookieServiceInfoFormat.parseFrom(bookieServiceInfo);
+ BookieServiceInfoFormat fmt = new BookieServiceInfoFormat();
+ fmt.parseFrom(bookieServiceInfo);
BookieServiceInfo bsi = new BookieServiceInfo();
- List<BookieServiceInfo.Endpoint> endpoints =
builder.getEndpointsList().stream()
- .map(e -> {
- BookieServiceInfo.Endpoint endpoint = new
BookieServiceInfo.Endpoint();
- endpoint.setId(e.getId());
- endpoint.setPort(e.getPort());
- endpoint.setHost(e.getHost());
- endpoint.setProtocol(e.getProtocol());
- endpoint.setAuth(e.getAuthList());
- endpoint.setExtensions(e.getExtensionsList());
- return endpoint;
- })
- .collect(Collectors.toList());
+ List<BookieServiceInfo.Endpoint> endpoints = new
ArrayList<>(fmt.getEndpointsCount());
+ for (int i = 0; i < fmt.getEndpointsCount(); i++) {
+ BookieServiceInfoFormat.Endpoint e = fmt.getEndpointAt(i);
+ BookieServiceInfo.Endpoint endpoint = new
BookieServiceInfo.Endpoint();
+ endpoint.setId(e.getId());
+ endpoint.setPort(e.getPort());
+ endpoint.setHost(e.getHost());
+ endpoint.setProtocol(e.getProtocol());
+ endpoint.setAuth(e.getAuthsList());
+ endpoint.setExtensions(e.getExtensionsList());
+ endpoints.add(endpoint);
+ }
bsi.setEndpoints(endpoints);
- bsi.setProperties(builder.getPropertiesMap());
+ Map<String, String> properties = new HashMap<>();
+ fmt.forEachProperties(properties::put);
+ bsi.setProperties(properties);
return bsi;
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
index 80a45c0dd7..84827125ca 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/discover/ZKRegistrationManager.java
@@ -27,7 +27,6 @@ import static
org.apache.bookkeeper.util.BookKeeperConstants.READONLY;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -37,7 +36,6 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
-import java.util.stream.Collectors;
import lombok.CustomLog;
import org.apache.bookkeeper.bookie.BookieException;
import org.apache.bookkeeper.bookie.BookieException.BookieIllegalOpException;
@@ -56,7 +54,7 @@ import org.apache.bookkeeper.meta.ZkLayoutManager;
import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieId;
-import org.apache.bookkeeper.proto.DataFormats.BookieServiceInfoFormat;
+import org.apache.bookkeeper.proto.BookieServiceInfoFormat;
import org.apache.bookkeeper.util.BookKeeperConstants;
import org.apache.bookkeeper.util.ZkUtils;
import org.apache.bookkeeper.versioning.LongVersion;
@@ -239,33 +237,18 @@ public class ZKRegistrationManager implements
RegistrationManager {
@VisibleForTesting
static byte[] serializeBookieServiceInfo(BookieServiceInfo
bookieServiceInfo) {
log.debug().attr("bookieServiceInfo",
bookieServiceInfo).log("serialize BookieServiceInfo");
- try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
- BookieServiceInfoFormat.Builder builder =
BookieServiceInfoFormat.newBuilder();
- List<BookieServiceInfoFormat.Endpoint> bsiEndpoints =
bookieServiceInfo.getEndpoints().stream()
- .map(e -> {
- return BookieServiceInfoFormat.Endpoint.newBuilder()
- .setId(e.getId())
- .setPort(e.getPort())
- .setHost(e.getHost())
- .setProtocol(e.getProtocol())
- .addAllAuth(e.getAuth())
- .addAllExtensions(e.getExtensions())
- .build();
- })
- .collect(Collectors.toList());
-
- builder.addAllEndpoints(bsiEndpoints);
- builder.putAllProperties(bookieServiceInfo.getProperties());
-
- builder.build().writeTo(os);
- return os.toByteArray();
- } catch (IOException err) {
- log.error()
- .exception(err)
- .attr("bookieServiceInfo", bookieServiceInfo)
- .log("Cannot serialize bookieServiceInfo");
- throw new RuntimeException(err);
+ BookieServiceInfoFormat fmt = new BookieServiceInfoFormat();
+ for (BookieServiceInfo.Endpoint e : bookieServiceInfo.getEndpoints()) {
+ BookieServiceInfoFormat.Endpoint endpoint = fmt.addEndpoint()
+ .setId(e.getId())
+ .setPort(e.getPort())
+ .setHost(e.getHost())
+ .setProtocol(e.getProtocol());
+ endpoint.addAllAuths(e.getAuth());
+ endpoint.addAllExtensions(e.getExtensions());
}
+ bookieServiceInfo.getProperties().forEach(fmt::putProperties);
+ return fmt.toByteArray();
}
private void doRegisterBookie(String regPath, BookieServiceInfo
bookieServiceInfo) throws BookieException {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerMetadataSerDe.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerMetadataSerDe.java
index bcbbf8473a..e8c24ba580 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerMetadataSerDe.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/LedgerMetadataSerDe.java
@@ -21,11 +21,12 @@ import static
com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static java.nio.charset.StandardCharsets.UTF_8;
-import com.google.protobuf.ByteString;
-import com.google.protobuf.TextFormat;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
@@ -35,10 +36,10 @@ import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Base64;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.stream.Collectors;
import lombok.CustomLog;
import org.apache.bookkeeper.client.LedgerMetadataBuilder;
import org.apache.bookkeeper.client.LedgerMetadataUtils;
@@ -46,7 +47,7 @@ import org.apache.bookkeeper.client.api.DigestType;
import org.apache.bookkeeper.client.api.LedgerMetadata;
import org.apache.bookkeeper.client.api.LedgerMetadata.State;
import org.apache.bookkeeper.net.BookieId;
-import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat;
/**
* Serialization and deserialization for LedgerMetadata.
@@ -152,61 +153,8 @@ public class LedgerMetadataSerDe {
private static byte[] serializeVersion3(LedgerMetadata metadata) throws
IOException {
try (ByteArrayOutputStream os = new ByteArrayOutputStream()) {
writeHeader(os, METADATA_FORMAT_VERSION_3);
- LedgerMetadataFormat.Builder builder =
LedgerMetadataFormat.newBuilder();
- builder.setQuorumSize(metadata.getWriteQuorumSize())
- .setAckQuorumSize(metadata.getAckQuorumSize())
- .setEnsembleSize(metadata.getEnsembleSize())
- .setLength(metadata.getLength())
- .setLastEntryId(metadata.getLastEntryId());
-
- switch (metadata.getState()) {
- case CLOSED:
- builder.setState(LedgerMetadataFormat.State.CLOSED);
- break;
- case IN_RECOVERY:
- builder.setState(LedgerMetadataFormat.State.IN_RECOVERY);
- break;
- case OPEN:
- builder.setState(LedgerMetadataFormat.State.OPEN);
- break;
- default:
- checkArgument(false,
- String.format("Unknown state %s for protobuf
serialization", metadata.getState()));
- break;
- }
-
- /** Hack to get around fact that ctime was never versioned
correctly */
- if (LedgerMetadataUtils.shouldStoreCtime(metadata)) {
- builder.setCtime(metadata.getCtime());
- }
-
-
-
builder.setDigestType(apiToProtoDigestType(metadata.getDigestType()));
-
- serializePassword(metadata.getPassword(), builder);
-
- Map<String, byte[]> customMetadata = metadata.getCustomMetadata();
- if (customMetadata.size() > 0) {
- LedgerMetadataFormat.cMetadataMapEntry.Builder
cMetadataBuilder =
- LedgerMetadataFormat.cMetadataMapEntry.newBuilder();
- for (Map.Entry<String, byte[]> entry :
customMetadata.entrySet()) {
-
cMetadataBuilder.setKey(entry.getKey()).setValue(ByteString.copyFrom(entry.getValue()));
- builder.addCustomMetadata(cMetadataBuilder.build());
- }
- }
-
- for (Map.Entry<Long, ? extends List<BookieId>> entry :
metadata.getAllEnsembles().entrySet()) {
- LedgerMetadataFormat.Segment.Builder segmentBuilder =
LedgerMetadataFormat.Segment.newBuilder();
- segmentBuilder.setFirstEntryId(entry.getKey());
- for (BookieId addr : entry.getValue()) {
- segmentBuilder.addEnsembleMember(addr.toString());
- }
- builder.addSegment(segmentBuilder.build());
- }
-
- builder.setCToken(metadata.getCToken());
-
- builder.build().writeDelimitedTo(os);
+ LedgerMetadataFormat fmt = buildLedgerMetadataFormat(metadata,
true);
+ writeDelimited(fmt, os);
return os.toByteArray();
}
}
@@ -221,62 +169,62 @@ public class LedgerMetadataSerDe {
* fields, and if this code was shared with version 3, it
would be easy to
* accidently add new fields and create BC issues.
**********************************************************************/
- LedgerMetadataFormat.Builder builder =
LedgerMetadataFormat.newBuilder();
- builder.setQuorumSize(metadata.getWriteQuorumSize())
- .setAckQuorumSize(metadata.getAckQuorumSize())
- .setEnsembleSize(metadata.getEnsembleSize())
- .setLength(metadata.getLength())
- .setLastEntryId(metadata.getLastEntryId());
-
- switch (metadata.getState()) {
- case CLOSED:
- builder.setState(LedgerMetadataFormat.State.CLOSED);
- break;
- case IN_RECOVERY:
- builder.setState(LedgerMetadataFormat.State.IN_RECOVERY);
- break;
- case OPEN:
- builder.setState(LedgerMetadataFormat.State.OPEN);
- break;
- default:
- checkArgument(false,
- String.format("Unknown state %s for protobuf
serialization", metadata.getState()));
- break;
- }
+ LedgerMetadataFormat fmt = buildLedgerMetadataFormat(metadata,
false);
+ writer.print(fmt.toTextFormat());
+ writer.flush();
+ }
+ return os.toByteArray();
+ }
+ }
- /** Hack to get around fact that ctime was never versioned
correctly */
- if (LedgerMetadataUtils.shouldStoreCtime(metadata)) {
- builder.setCtime(metadata.getCtime());
- }
+ private static LedgerMetadataFormat
buildLedgerMetadataFormat(LedgerMetadata metadata, boolean includeCToken) {
+ LedgerMetadataFormat fmt = new LedgerMetadataFormat()
+ .setQuorumSize(metadata.getWriteQuorumSize())
+ .setAckQuorumSize(metadata.getAckQuorumSize())
+ .setEnsembleSize(metadata.getEnsembleSize())
+ .setLength(metadata.getLength())
+ .setLastEntryId(metadata.getLastEntryId());
-
builder.setDigestType(apiToProtoDigestType(metadata.getDigestType()));
- serializePassword(metadata.getPassword(), builder);
+ switch (metadata.getState()) {
+ case CLOSED:
+ fmt.setState(LedgerMetadataFormat.State.CLOSED);
+ break;
+ case IN_RECOVERY:
+ fmt.setState(LedgerMetadataFormat.State.IN_RECOVERY);
+ break;
+ case OPEN:
+ fmt.setState(LedgerMetadataFormat.State.OPEN);
+ break;
+ default:
+ checkArgument(false,
+ String.format("Unknown state %s for protobuf
serialization", metadata.getState()));
+ break;
+ }
- Map<String, byte[]> customMetadata =
metadata.getCustomMetadata();
- if (customMetadata.size() > 0) {
- LedgerMetadataFormat.cMetadataMapEntry.Builder
cMetadataBuilder =
- LedgerMetadataFormat.cMetadataMapEntry.newBuilder();
- for (Map.Entry<String, byte[]> entry :
customMetadata.entrySet()) {
-
cMetadataBuilder.setKey(entry.getKey()).setValue(ByteString.copyFrom(entry.getValue()));
- builder.addCustomMetadata(cMetadataBuilder.build());
- }
- }
+ /** Hack to get around fact that ctime was never versioned correctly */
+ if (LedgerMetadataUtils.shouldStoreCtime(metadata)) {
+ fmt.setCtime(metadata.getCtime());
+ }
- for (Map.Entry<Long, ? extends List<BookieId>> entry :
- metadata.getAllEnsembles().entrySet()) {
- LedgerMetadataFormat.Segment.Builder segmentBuilder =
LedgerMetadataFormat.Segment.newBuilder();
- segmentBuilder.setFirstEntryId(entry.getKey());
- for (BookieId addr : entry.getValue()) {
- segmentBuilder.addEnsembleMember(addr.toString());
- }
- builder.addSegment(segmentBuilder.build());
- }
+ fmt.setDigestType(apiToProtoDigestType(metadata.getDigestType()));
+ serializePassword(metadata.getPassword(), fmt);
- TextFormat.printer().print(builder.build(), writer);
- writer.flush();
+ for (Map.Entry<String, byte[]> entry :
metadata.getCustomMetadata().entrySet()) {
+
fmt.addCustomMetadata().setKey(entry.getKey()).setValue(entry.getValue());
+ }
+
+ for (Map.Entry<Long, ? extends List<BookieId>> entry :
metadata.getAllEnsembles().entrySet()) {
+ LedgerMetadataFormat.Segment seg = fmt.addSegment();
+ seg.setFirstEntryId(entry.getKey());
+ for (BookieId addr : entry.getValue()) {
+ seg.addEnsembleMember(addr.toString());
}
- return os.toByteArray();
}
+
+ if (includeCToken) {
+ fmt.setCToken(metadata.getCToken());
+ }
+ return fmt;
}
private static byte[] serializeVersion1(LedgerMetadata metadata) throws
IOException {
@@ -314,11 +262,11 @@ public class LedgerMetadataSerDe {
}
}
- private static void serializePassword(byte[] password,
LedgerMetadataFormat.Builder builder) {
+ private static void serializePassword(byte[] password,
LedgerMetadataFormat fmt) {
if (password == null || password.length == 0) {
- builder.setPassword(ByteString.EMPTY);
+ fmt.setPassword(new byte[0]);
} else {
- builder.setPassword(ByteString.copyFrom(password));
+ fmt.setPassword(password);
}
}
@@ -368,9 +316,8 @@ public class LedgerMetadataSerDe {
LedgerMetadataBuilder builder = LedgerMetadataBuilder.create()
.withId(ledgerId)
.withMetadataFormatVersion(METADATA_FORMAT_VERSION_3);
- LedgerMetadataFormat.Builder formatBuilder =
LedgerMetadataFormat.newBuilder();
- formatBuilder.mergeDelimitedFrom(is);
- LedgerMetadataFormat data = formatBuilder.build();
+ LedgerMetadataFormat data = new LedgerMetadataFormat();
+ readDelimited(is, data);
decodeFormat(data, builder);
if (data.hasCtime()) {
builder.storingCreationTime(true);
@@ -386,11 +333,8 @@ public class LedgerMetadataSerDe {
.withId(ledgerId)
.withMetadataFormatVersion(METADATA_FORMAT_VERSION_2);
- LedgerMetadataFormat.Builder formatBuilder =
LedgerMetadataFormat.newBuilder();
- try (InputStreamReader reader = new InputStreamReader(is,
UTF_8.name())) {
- TextFormat.merge(reader, formatBuilder);
- }
- LedgerMetadataFormat data = formatBuilder.build();
+ LedgerMetadataFormat data = new LedgerMetadataFormat();
+ data.parseFromTextFormat(is.readAllBytes());
decodeFormat(data, builder);
if (data.hasCtime()) {
// 'storingCreationTime' is only ever taken into account for
serializing version 2
@@ -421,22 +365,26 @@ public class LedgerMetadataSerDe {
}
if (data.hasPassword()) {
- builder.withPassword(data.getPassword().toByteArray())
+ builder.withPassword(data.getPassword())
.withDigestType(protoToApiDigestType(data.getDigestType()));
}
- for (LedgerMetadataFormat.Segment s : data.getSegmentList()) {
- List<BookieId> addrs = new ArrayList<>();
- for (String addr : s.getEnsembleMemberList()) {
- addrs.add(BookieId.parse(addr));
+ for (int i = 0; i < data.getSegmentsCount(); i++) {
+ LedgerMetadataFormat.Segment s = data.getSegmentAt(i);
+ List<BookieId> addrs = new
ArrayList<>(s.getEnsembleMembersCount());
+ for (int j = 0; j < s.getEnsembleMembersCount(); j++) {
+ addrs.add(BookieId.parse(s.getEnsembleMemberAt(j)));
}
builder.newEnsembleEntry(s.getFirstEntryId(), addrs);
}
- if (data.getCustomMetadataCount() > 0) {
-
builder.withCustomMetadata(data.getCustomMetadataList().stream().collect(
- Collectors.toMap(e ->
e.getKey(),
- e ->
e.getValue().toByteArray())));
+ if (data.getCustomMetadatasCount() > 0) {
+ Map<String, byte[]> customMetadata = new HashMap<>();
+ for (int i = 0; i < data.getCustomMetadatasCount(); i++) {
+ LedgerMetadataFormat.cMetadataMapEntry e =
data.getCustomMetadataAt(i);
+ customMetadata.put(e.getKey(), e.getValue());
+ }
+ builder.withCustomMetadata(customMetadata);
}
if (data.hasCToken()) {
@@ -512,4 +460,60 @@ public class LedgerMetadataSerDe {
throw new IllegalArgumentException("Unable to convert digest type
" + digestType);
}
}
+
+ /** Write a length-prefixed message in protobuf-compatible delimited
format. */
+ private static void writeDelimited(LedgerMetadataFormat msg, OutputStream
os) throws IOException {
+ int size = msg.getSerializedSize();
+ writeRawVarint32(size, os);
+ ByteBuf buf = Unpooled.buffer(size);
+ try {
+ msg.writeTo(buf);
+ os.write(buf.array(), buf.arrayOffset() + buf.readerIndex(),
buf.readableBytes());
+ } finally {
+ buf.release();
+ }
+ }
+
+ /** Read a length-prefixed message in protobuf-compatible delimited
format. */
+ private static void readDelimited(InputStream is, LedgerMetadataFormat
msg) throws IOException {
+ int size = readRawVarint32(is);
+ byte[] data = new byte[size];
+ int read = 0;
+ while (read < size) {
+ int n = is.read(data, read, size - read);
+ if (n < 0) {
+ throw new EOFException("Unexpected EOF reading delimited
message");
+ }
+ read += n;
+ }
+ msg.parseFrom(data);
+ }
+
+ private static void writeRawVarint32(int value, OutputStream os) throws
IOException {
+ while (true) {
+ if ((value & ~0x7F) == 0) {
+ os.write(value);
+ return;
+ }
+ os.write((value & 0x7F) | 0x80);
+ value >>>= 7;
+ }
+ }
+
+ private static int readRawVarint32(InputStream is) throws IOException {
+ int result = 0;
+ int shift = 0;
+ while (shift < 32) {
+ int b = is.read();
+ if (b == -1) {
+ throw new EOFException("Unexpected EOF reading varint");
+ }
+ result |= (b & 0x7F) << shift;
+ if ((b & 0x80) == 0) {
+ return result;
+ }
+ shift += 7;
+ }
+ throw new IOException("Malformed varint32");
+ }
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerAuditorManager.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerAuditorManager.java
index 768b2663cb..e174cc7b4f 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerAuditorManager.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerAuditorManager.java
@@ -18,10 +18,8 @@
package org.apache.bookkeeper.meta;
import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.apache.bookkeeper.proto.DataFormats.AuditorVoteFormat;
import static
org.apache.bookkeeper.replication.ReplicationStats.ELECTION_ATTEMPTS;
-import com.google.protobuf.TextFormat;
import java.io.IOException;
import java.io.Serializable;
import java.util.Collections;
@@ -33,6 +31,7 @@ import lombok.CustomLog;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.proto.AuditorVoteFormat;
import org.apache.bookkeeper.stats.Counter;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.stats.annotations.StatsDoc;
@@ -112,11 +111,11 @@ public class ZkLedgerAuditorManager implements
LedgerAuditorManager {
// We have been elected as the auditor
// update the auditor bookie id in the election path. This
is
// done for debugging purpose
- AuditorVoteFormat.Builder builder =
AuditorVoteFormat.newBuilder()
+ AuditorVoteFormat fmt = new AuditorVoteFormat()
.setBookieId(bookieId);
zkc.setData(getVotePath(""),
- builder.build().toString().getBytes(UTF_8), -1);
+ fmt.toTextFormat().getBytes(UTF_8), -1);
return;
} else {
// If not an auditor, will be watching to my predecessor
and
@@ -162,9 +161,8 @@ public class ZkLedgerAuditorManager implements
LedgerAuditorManager {
String ledger = electionRoot + "/" + children.get(AUDITOR_INDEX);
byte[] data = zkc.getData(ledger, false, null);
- AuditorVoteFormat.Builder builder = AuditorVoteFormat.newBuilder();
- TextFormat.merge(new String(data, UTF_8), builder);
- AuditorVoteFormat v = builder.build();
+ AuditorVoteFormat v = new AuditorVoteFormat();
+ v.parseFromTextFormat(data);
return BookieId.parse(v.getBookieId());
} catch (KeeperException e) {
throw new IOException(e);
@@ -191,13 +189,13 @@ public class ZkLedgerAuditorManager implements
LedgerAuditorManager {
private void createMyVote(String bookieId) throws IOException,
InterruptedException {
List<ACL> zkAcls = ZkUtils.getACLs(conf);
- AuditorVoteFormat.Builder builder = AuditorVoteFormat.newBuilder()
+ AuditorVoteFormat fmt = new AuditorVoteFormat()
.setBookieId(bookieId);
try {
if (null == myVote || null == zkc.exists(myVote, false)) {
myVote = zkc.create(getVotePath(PATH_SEPARATOR + VOTE_PREFIX),
- builder.build().toString().getBytes(UTF_8), zkAcls,
+ fmt.toTextFormat().getBytes(UTF_8), zkAcls,
CreateMode.EPHEMERAL_SEQUENTIAL);
}
} catch (KeeperException e) {
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java
index 7bc6ccf9b8..00e0d77d43 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/meta/ZkLedgerUnderreplicationManager.java
@@ -22,9 +22,6 @@ import static java.nio.charset.StandardCharsets.UTF_8;
import com.google.common.base.Joiner;
import com.google.common.util.concurrent.RateLimiter;
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.google.protobuf.TextFormat;
-import com.google.protobuf.TextFormat.ParseException;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collection;
@@ -47,12 +44,12 @@ import org.apache.bookkeeper.conf.AbstractConfiguration;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.DNS;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
-import org.apache.bookkeeper.proto.DataFormats.CheckAllLedgersFormat;
-import org.apache.bookkeeper.proto.DataFormats.LedgerRereplicationLayoutFormat;
-import org.apache.bookkeeper.proto.DataFormats.LockDataFormat;
-import org.apache.bookkeeper.proto.DataFormats.PlacementPolicyCheckFormat;
-import org.apache.bookkeeper.proto.DataFormats.ReplicasCheckFormat;
-import org.apache.bookkeeper.proto.DataFormats.UnderreplicatedLedgerFormat;
+import org.apache.bookkeeper.proto.CheckAllLedgersFormat;
+import org.apache.bookkeeper.proto.LedgerRereplicationLayoutFormat;
+import org.apache.bookkeeper.proto.LockDataFormat;
+import org.apache.bookkeeper.proto.PlacementPolicyCheckFormat;
+import org.apache.bookkeeper.proto.ReplicasCheckFormat;
+import org.apache.bookkeeper.proto.UnderreplicatedLedgerFormat;
import org.apache.bookkeeper.replication.ReplicationEnableCb;
import org.apache.bookkeeper.replication.ReplicationException;
import
org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
@@ -173,14 +170,14 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
}
public static byte[] getLockData() {
- LockDataFormat.Builder lockDataBuilder = LockDataFormat.newBuilder();
+ LockDataFormat lockData = new LockDataFormat();
try {
- lockDataBuilder.setBookieId(DNS.getDefaultHost("default"));
+ lockData.setBookieId(DNS.getDefaultHost("default"));
} catch (UnknownHostException uhe) {
// if we cant get the address, ignore. it's optional
// in the data structure in any case
}
- return lockDataBuilder.build().toString().getBytes(UTF_8);
+ return lockData.toTextFormat().getBytes(UTF_8);
}
private void checkLayout()
@@ -195,10 +192,10 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
}
while (true) {
if (zkc.exists(layoutZNode, false) == null) {
- LedgerRereplicationLayoutFormat.Builder builder =
LedgerRereplicationLayoutFormat.newBuilder();
- builder.setType(LAYOUT).setVersion(LAYOUT_VERSION);
+ LedgerRereplicationLayoutFormat layout = new
LedgerRereplicationLayoutFormat();
+ layout.setType(LAYOUT).setVersion(LAYOUT_VERSION);
try {
- zkc.create(layoutZNode,
builder.build().toString().getBytes(UTF_8),
+ zkc.create(layoutZNode,
layout.toTextFormat().getBytes(UTF_8),
zkAcls, CreateMode.PERSISTENT);
break;
} catch (KeeperException.NodeExistsException nne) {
@@ -207,17 +204,16 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
} else {
byte[] layoutData = zkc.getData(layoutZNode, false, null);
- LedgerRereplicationLayoutFormat.Builder builder =
LedgerRereplicationLayoutFormat.newBuilder();
+ LedgerRereplicationLayoutFormat layout = new
LedgerRereplicationLayoutFormat();
try {
- TextFormat.merge(new String(layoutData, UTF_8), builder);
- LedgerRereplicationLayoutFormat layout = builder.build();
+ layout.parseFromTextFormat(layoutData);
if (!layout.getType().equals(LAYOUT)
|| layout.getVersion() != LAYOUT_VERSION) {
throw new ReplicationException.CompatibilityException(
"Incompatible layout found (" + LAYOUT + ":" +
LAYOUT_VERSION + ")");
}
- } catch (TextFormat.ParseException pe) {
+ } catch (RuntimeException pe) {
throw new ReplicationException.CompatibilityException(
"Invalid data found", pe);
}
@@ -276,7 +272,7 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
throws ReplicationException.UnavailableException {
try {
String znode = getUrLedgerZnode(ledgerId);
- UnderreplicatedLedgerFormat.Builder builder =
UnderreplicatedLedgerFormat.newBuilder();
+ UnderreplicatedLedgerFormat fmt = new
UnderreplicatedLedgerFormat();
byte[] data = null;
try {
data = zkc.getData(znode, false, null);
@@ -284,11 +280,10 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
log.debug().attr("ledgerId", ledgerId).log("Ledger is not
marked underreplicated");
return null;
}
- TextFormat.merge(new String(data, UTF_8), builder);
- UnderreplicatedLedgerFormat underreplicatedLedgerFormat =
builder.build();
+ fmt.parseFromTextFormat(data);
UnderreplicatedLedger underreplicatedLedger = new
UnderreplicatedLedger(ledgerId);
- List<String> replicaList =
underreplicatedLedgerFormat.getReplicaList();
- long ctime = (underreplicatedLedgerFormat.hasCtime() ?
underreplicatedLedgerFormat.getCtime()
+ List<String> replicaList = fmt.getReplicasList();
+ long ctime = (fmt.hasCtime() ? fmt.getCtime()
: UnderreplicatedLedger.UNASSIGNED_CTIME);
underreplicatedLedger.setCtime(ctime);
underreplicatedLedger.setReplicaList(replicaList);
@@ -298,7 +293,7 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted
while connecting zookeeper", ie);
- } catch (TextFormat.ParseException pe) {
+ } catch (RuntimeException pe) {
throw new ReplicationException.UnavailableException("Error parsing
proto message", pe);
}
}
@@ -320,12 +315,12 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
final Collection<String>
missingReplicas,
final List<ACL> zkAcls,
final
CompletableFuture<Void> finalFuture) {
- final UnderreplicatedLedgerFormat.Builder builder =
UnderreplicatedLedgerFormat.newBuilder();
+ final UnderreplicatedLedgerFormat fmt = new
UnderreplicatedLedgerFormat();
if (conf.getStoreSystemTimeAsLedgerUnderreplicatedMarkTime()) {
- builder.setCtime(System.currentTimeMillis());
+ fmt.setCtime(System.currentTimeMillis());
}
- missingReplicas.forEach(builder::addReplica);
- final byte[] urLedgerData = builder.build().toString().getBytes(UTF_8);
+ missingReplicas.forEach(fmt::addReplica);
+ final byte[] urLedgerData = fmt.toTextFormat().getBytes(UTF_8);
ZkUtils.asyncCreateFullPathOptimistic(
zkc, znode, urLedgerData, zkAcls, CreateMode.PERSISTENT,
(rc, path, ctx, name) -> {
@@ -349,23 +344,22 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
zkc.getData(znode, false, (getRc, getPath, getCtx,
existingUrLedgerData, getStat) -> {
if (Code.OK.intValue() == getRc) {
// deserialize existing underreplicated ledger data
- final UnderreplicatedLedgerFormat.Builder builder =
UnderreplicatedLedgerFormat.newBuilder();
+ final UnderreplicatedLedgerFormat fmt = new
UnderreplicatedLedgerFormat();
try {
- TextFormat.merge(new String(existingUrLedgerData, UTF_8),
builder);
- } catch (ParseException e) {
+ fmt.parseFromTextFormat(existingUrLedgerData);
+ } catch (RuntimeException e) {
// corrupted metadata in zookeeper
FutureUtils.completeExceptionally(finalFuture,
new ReplicationException.UnavailableException(
"Invalid underreplicated ledger data for ledger "
+ znode, e));
return;
}
- UnderreplicatedLedgerFormat existingUrLedgerFormat =
builder.build();
boolean replicaAdded = false;
for (String missingReplica : missingReplicas) {
- if
(existingUrLedgerFormat.getReplicaList().contains(missingReplica)) {
+ if (fmt.getReplicasList().contains(missingReplica)) {
continue;
} else {
- builder.addReplica(missingReplica);
+ fmt.addReplica(missingReplica);
replicaAdded = true;
}
}
@@ -374,9 +368,9 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
return;
}
if (conf.getStoreSystemTimeAsLedgerUnderreplicatedMarkTime()) {
- builder.setCtime(System.currentTimeMillis());
+ fmt.setCtime(System.currentTimeMillis());
}
- final byte[] newUrLedgerData =
builder.build().toString().getBytes(UTF_8);
+ final byte[] newUrLedgerData =
fmt.toTextFormat().getBytes(UTF_8);
zkc.setData(znode, newUrLedgerData, getStat.getVersion(),
(setRc, setPath, setCtx, setStat) -> {
if (Code.OK.intValue() == setRc) {
FutureUtils.complete(finalFuture, null);
@@ -917,9 +911,8 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
String replicationWorkerId = null;
try {
byte[] lockData = zkc.getData(getUrLedgerLockZnode(urLockPath,
ledgerId), false, null);
- LockDataFormat.Builder lockDataBuilder =
LockDataFormat.newBuilder();
- TextFormat.merge(new String(lockData, UTF_8), lockDataBuilder);
- LockDataFormat lock = lockDataBuilder.build();
+ LockDataFormat lock = new LockDataFormat();
+ lock.parseFromTextFormat(lockData);
replicationWorkerId = lock.getBookieId();
} catch (KeeperException.NoNodeException e) {
// this is ok.
@@ -931,7 +924,7 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
log.error().exception(e).log("Got interrupted while getting
ReplicationWorkerId rereplicating Ledger");
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted
while contacting zookeeper", e);
- } catch (ParseException e) {
+ } catch (RuntimeException e) {
log.error().exception(e).log("Error while parsing ZK data of
lock");
throw new ReplicationException.UnavailableException("Error while
parsing ZK data of lock", e);
}
@@ -943,9 +936,9 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
log.debug("setCheckAllLedgersCTime");
try {
List<ACL> zkAcls = ZkUtils.getACLs(conf);
- CheckAllLedgersFormat.Builder builder =
CheckAllLedgersFormat.newBuilder();
- builder.setCheckAllLedgersCTime(checkAllLedgersCTime);
- byte[] checkAllLedgersFormatByteArray =
builder.build().toByteArray();
+ CheckAllLedgersFormat fmt = new CheckAllLedgersFormat();
+ fmt.setCheckAllLedgersCTime(checkAllLedgersCTime);
+ byte[] checkAllLedgersFormatByteArray = fmt.toByteArray();
if (zkc.exists(checkAllLedgersCtimeZnode, false) != null) {
zkc.setData(checkAllLedgersCtimeZnode,
checkAllLedgersFormatByteArray, -1);
} else {
@@ -964,7 +957,8 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
log.debug("getCheckAllLedgersCTime");
try {
byte[] data = zkc.getData(checkAllLedgersCtimeZnode, false, null);
- CheckAllLedgersFormat checkAllLedgersFormat =
CheckAllLedgersFormat.parseFrom(data);
+ CheckAllLedgersFormat checkAllLedgersFormat = new
CheckAllLedgersFormat();
+ checkAllLedgersFormat.parseFrom(data);
return checkAllLedgersFormat.hasCheckAllLedgersCTime() ?
checkAllLedgersFormat.getCheckAllLedgersCTime()
: -1;
} catch (KeeperException.NoNodeException ne) {
@@ -975,7 +969,7 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted
while contacting zookeeper", ie);
- } catch (InvalidProtocolBufferException ipbe) {
+ } catch (RuntimeException ipbe) {
throw new ReplicationException.UnavailableException("Error while
parsing ZK protobuf binary data", ipbe);
}
}
@@ -985,9 +979,9 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
log.debug("setPlacementPolicyCheckCTime");
try {
List<ACL> zkAcls = ZkUtils.getACLs(conf);
- PlacementPolicyCheckFormat.Builder builder =
PlacementPolicyCheckFormat.newBuilder();
- builder.setPlacementPolicyCheckCTime(placementPolicyCheckCTime);
- byte[] placementPolicyCheckFormatByteArray =
builder.build().toByteArray();
+ PlacementPolicyCheckFormat fmt = new PlacementPolicyCheckFormat();
+ fmt.setPlacementPolicyCheckCTime(placementPolicyCheckCTime);
+ byte[] placementPolicyCheckFormatByteArray = fmt.toByteArray();
if (zkc.exists(placementPolicyCheckCtimeZnode, false) != null) {
zkc.setData(placementPolicyCheckCtimeZnode,
placementPolicyCheckFormatByteArray, -1);
} else {
@@ -1007,7 +1001,8 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
log.debug("getPlacementPolicyCheckCTime");
try {
byte[] data = zkc.getData(placementPolicyCheckCtimeZnode, false,
null);
- PlacementPolicyCheckFormat placementPolicyCheckFormat =
PlacementPolicyCheckFormat.parseFrom(data);
+ PlacementPolicyCheckFormat placementPolicyCheckFormat = new
PlacementPolicyCheckFormat();
+ placementPolicyCheckFormat.parseFrom(data);
return placementPolicyCheckFormat.hasPlacementPolicyCheckCTime()
?
placementPolicyCheckFormat.getPlacementPolicyCheckCTime() : -1;
} catch (KeeperException.NoNodeException ne) {
@@ -1018,7 +1013,7 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted
while contacting zookeeper", ie);
- } catch (InvalidProtocolBufferException ipbe) {
+ } catch (RuntimeException ipbe) {
throw new ReplicationException.UnavailableException("Error while
parsing ZK protobuf binary data", ipbe);
}
}
@@ -1027,9 +1022,9 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
public void setReplicasCheckCTime(long replicasCheckCTime) throws
UnavailableException {
try {
List<ACL> zkAcls = ZkUtils.getACLs(conf);
- ReplicasCheckFormat.Builder builder =
ReplicasCheckFormat.newBuilder();
- builder.setReplicasCheckCTime(replicasCheckCTime);
- byte[] replicasCheckFormatByteArray =
builder.build().toByteArray();
+ ReplicasCheckFormat fmt = new ReplicasCheckFormat();
+ fmt.setReplicasCheckCTime(replicasCheckCTime);
+ byte[] replicasCheckFormatByteArray = fmt.toByteArray();
if (zkc.exists(replicasCheckCtimeZnode, false) != null) {
zkc.setData(replicasCheckCtimeZnode,
replicasCheckFormatByteArray, -1);
} else {
@@ -1048,7 +1043,8 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
public long getReplicasCheckCTime() throws UnavailableException {
try {
byte[] data = zkc.getData(replicasCheckCtimeZnode, false, null);
- ReplicasCheckFormat replicasCheckFormat =
ReplicasCheckFormat.parseFrom(data);
+ ReplicasCheckFormat replicasCheckFormat = new
ReplicasCheckFormat();
+ replicasCheckFormat.parseFrom(data);
log.debug("getReplicasCheckCTime completed successfully");
return replicasCheckFormat.hasReplicasCheckCTime() ?
replicasCheckFormat.getReplicasCheckCTime() : -1;
} catch (KeeperException.NoNodeException ne) {
@@ -1059,7 +1055,7 @@ public class ZkLedgerUnderreplicationManager implements
LedgerUnderreplicationMa
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
throw new ReplicationException.UnavailableException("Interrupted
while contacting zookeeper", ie);
- } catch (InvalidProtocolBufferException ipbe) {
+ } catch (RuntimeException ipbe) {
throw new ReplicationException.UnavailableException("Error while
parsing ZK protobuf binary data", ipbe);
}
}
diff --git
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java
index 114df52de8..b84492514a 100644
---
a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java
+++
b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java
@@ -31,7 +31,7 @@ import
org.apache.bookkeeper.client.BKException.BKDigestMatchException;
import org.apache.bookkeeper.client.LedgerHandle;
import org.apache.bookkeeper.proto.BookieProtoEncoding;
import org.apache.bookkeeper.proto.BookieProtocol;
-import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.DigestType;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat.DigestType;
import org.apache.bookkeeper.util.ByteBufList;
import org.apache.bookkeeper.util.ByteBufVisitor;
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
index 7269265d5a..49e9645a2f 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/BookieInitializationTest.java
@@ -95,7 +95,7 @@ import org.apache.bookkeeper.meta.zk.ZKMetadataBookieDriver;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.proto.BookieServer;
-import org.apache.bookkeeper.proto.DataFormats.BookieServiceInfoFormat;
+import org.apache.bookkeeper.proto.BookieServiceInfoFormat;
import org.apache.bookkeeper.replication.AutoRecoveryMain;
import org.apache.bookkeeper.replication.ReplicationStats;
import org.apache.bookkeeper.server.Main;
@@ -550,8 +550,9 @@ public class BookieInitializationTest extends
BookKeeperClusterTestCase {
byte[] bkRegNodeData = zkc.getData(bkRegPath, null, null);
assertFalse("Bookie service info not written", bkRegNodeData == null
|| bkRegNodeData.length == 0);
- BookieServiceInfoFormat serializedBookieServiceInfo =
BookieServiceInfoFormat.parseFrom(bkRegNodeData);
- BookieServiceInfoFormat.Endpoint serializedEndpoint =
serializedBookieServiceInfo.getEndpoints(0);
+ BookieServiceInfoFormat serializedBookieServiceInfo = new
BookieServiceInfoFormat();
+ serializedBookieServiceInfo.parseFrom(bkRegNodeData);
+ BookieServiceInfoFormat.Endpoint serializedEndpoint =
serializedBookieServiceInfo.getEndpointAt(0);
assertNotNull("Serialized Bookie endpoint not found",
serializedEndpoint);
assertEquals(endpoint.getId(), serializedEndpoint.getId());
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java
index 3f8af53c13..8c66401158 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ClientUtil.java
@@ -26,7 +26,7 @@ import java.security.GeneralSecurityException;
import java.util.function.Function;
import org.apache.bookkeeper.client.api.LedgerMetadata;
import org.apache.bookkeeper.meta.LedgerManager;
-import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.DigestType;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat.DigestType;
import org.apache.bookkeeper.proto.MockBookieClient;
import org.apache.bookkeeper.proto.checksum.DigestManager;
import org.apache.bookkeeper.versioning.Versioned;
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedOpTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedOpTest.java
index 954bdccca8..e51ffd3ff7 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedOpTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/ReadLastConfirmedOpTest.java
@@ -27,7 +27,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.bookkeeper.common.util.OrderedExecutor;
import org.apache.bookkeeper.net.BookieId;
import org.apache.bookkeeper.net.BookieSocketAddress;
-import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.DigestType;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat.DigestType;
import org.apache.bookkeeper.proto.MockBookieClient;
import org.apache.bookkeeper.proto.checksum.DigestManager;
import org.junit.After;
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookies.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookies.java
index ac338b9757..0cb3ad954f 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookies.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/proto/MockBookies.java
@@ -82,7 +82,7 @@ public class MockBookies {
public ByteBuf generateEntry(long ledgerId, long entryId, long lac) throws
Exception {
DigestManager digestManager = DigestManager.instantiate(ledgerId, new
byte[0],
- DataFormats.LedgerMetadataFormat.DigestType.CRC32C,
+ LedgerMetadataFormat.DigestType.CRC32C,
UnpooledByteBufAllocator.DEFAULT, false);
return ByteBufList.coalesce((ByteBufList)
digestManager.computeDigestAndPackageForSending(
entryId, lac, 0, Unpooled.buffer(10), new byte[20], 0));
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java
index 3e418226e6..67d68b3353 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestLedgerUnderreplicationManager.java
@@ -26,7 +26,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import com.google.protobuf.TextFormat;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Collection;
@@ -53,7 +52,7 @@ import org.apache.bookkeeper.meta.ZkLayoutManager;
import org.apache.bookkeeper.meta.ZkLedgerUnderreplicationManager;
import org.apache.bookkeeper.meta.zk.ZKMetadataDriverBase;
import org.apache.bookkeeper.net.DNS;
-import org.apache.bookkeeper.proto.DataFormats.UnderreplicatedLedgerFormat;
+import org.apache.bookkeeper.proto.UnderreplicatedLedgerFormat;
import
org.apache.bookkeeper.replication.ReplicationException.UnavailableException;
import org.apache.bookkeeper.test.ZooKeeperUtil;
import org.apache.bookkeeper.util.BookKeeperConstants;
@@ -430,12 +429,11 @@ public class TestLedgerUnderreplicationManager {
m2.markLedgerUnderreplicated(ledgerA, missingReplica1);
// verify duplicate missing replica
- UnderreplicatedLedgerFormat.Builder builderA =
UnderreplicatedLedgerFormat
- .newBuilder();
+ UnderreplicatedLedgerFormat fmtA = new UnderreplicatedLedgerFormat();
String znode = getUrLedgerZnode(ledgerA);
byte[] data = zkc1.getData(znode, false, null);
- TextFormat.merge(new String(data, Charset.forName("UTF-8")), builderA);
- List<String> replicaList = builderA.getReplicaList();
+ fmtA.parseFromTextFormat(new String(data, Charset.forName("UTF-8")));
+ List<String> replicaList = fmtA.getReplicasList();
assertEquals("Published duplicate missing replica : " + replicaList, 1,
replicaList.size());
assertTrue("Published duplicate missing replica : " + replicaList,
@@ -833,12 +831,11 @@ public class TestLedgerUnderreplicationManager {
}
String urLedgerA = getData(znodeA);
- UnderreplicatedLedgerFormat.Builder builderA =
UnderreplicatedLedgerFormat
- .newBuilder();
+ UnderreplicatedLedgerFormat fmtA = new UnderreplicatedLedgerFormat();
for (String replica : missingReplica) {
- builderA.addReplica(replica);
+ fmtA.addReplica(replica);
}
- List<String> replicaList = builderA.getReplicaList();
+ List<String> replicaList = fmtA.getReplicasList();
for (String replica : missingReplica) {
assertTrue("UrLedger:" + urLedgerA
diff --git
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
index d8ab2f5989..e9928a01c8 100644
---
a/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
+++
b/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BookieClientTest.java
@@ -70,7 +70,7 @@ import
org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GetBookieInfoCall
import
org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.ReadEntryCallback;
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.WriteCallback;
import org.apache.bookkeeper.proto.BookkeeperProtocol;
-import org.apache.bookkeeper.proto.DataFormats;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat;
import org.apache.bookkeeper.proto.PerChannelBookieClient;
import org.apache.bookkeeper.proto.PerChannelBookieClientPool;
import org.apache.bookkeeper.proto.checksum.DigestManager;
@@ -381,7 +381,7 @@ public class BookieClientTest {
byte[] passwd = new byte[20];
Arrays.fill(passwd, (byte) 'a');
DigestManager digestManager = DigestManager.instantiate(1, passwd,
- DataFormats.LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
+ LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
byte[] masterKey = DigestManager.generateMasterKey(passwd);
final int entries = 10;
@@ -443,7 +443,7 @@ public class BookieClientTest {
byte[] passwd = new byte[20];
Arrays.fill(passwd, (byte) 'a');
DigestManager digestManager = DigestManager.instantiate(1, passwd,
- DataFormats.LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
+ LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
byte[] masterKey = DigestManager.generateMasterKey(passwd);
final int entries = 10;
@@ -509,7 +509,7 @@ public class BookieClientTest {
byte[] passwd = new byte[20];
Arrays.fill(passwd, (byte) 'a');
DigestManager digestManager = DigestManager.instantiate(1, passwd,
- DataFormats.LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
+ LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
byte[] masterKey = DigestManager.generateMasterKey(passwd);
final int entries = 10;
@@ -560,7 +560,7 @@ public class BookieClientTest {
byte[] passwd = new byte[20];
Arrays.fill(passwd, (byte) 'a');
DigestManager digestManager = DigestManager.instantiate(1, passwd,
- DataFormats.LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
+ LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
byte[] masterKey = DigestManager.generateMasterKey(passwd);
byte[] kbData = new byte[1024];
for (int i = 0; i < 1024; i++) {
@@ -627,7 +627,7 @@ public class BookieClientTest {
byte[] passwd = new byte[20];
Arrays.fill(passwd, (byte) 'a');
DigestManager digestManager = DigestManager.instantiate(1, passwd,
- DataFormats.LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
+ LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
byte[] masterKey = DigestManager.generateMasterKey(passwd);
byte[] kbData = new byte[1024];
for (int i = 0; i < 1024; i++) {
@@ -697,7 +697,7 @@ public class BookieClientTest {
byte[] passwd = new byte[20];
Arrays.fill(passwd, (byte) 'a');
DigestManager digestManager = DigestManager.instantiate(1, passwd,
- DataFormats.LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
+ LedgerMetadataFormat.DigestType.CRC32C,
ByteBufAllocator.DEFAULT, true);
byte[] masterKey = DigestManager.generateMasterKey(passwd);
byte[] kbData = new byte[1024];
for (int i = 0; i < 1024; i++) {
diff --git a/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml
b/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml
index ea2b65e447..326cfb36ee 100644
--- a/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml
+++ b/buildtools/src/main/resources/bookkeeper/findbugsExclude.xml
@@ -40,15 +40,15 @@
<!-- bookkeeper-proto -->
<Match>
<!-- generated code, we can't be held responsible for findbugs in it //-->
- <Class name="~org\.apache\.bookkeeper\.proto\.DataFormats.*" />
+ <Class name="~org\.apache\.bookkeeper\.proto\.BookkeeperProtocol.*" />
</Match>
+ <!-- lightproto-generated classes from DataFormats.proto -->
<Match>
- <!-- generated code, we can't be held responsible for findbugs in it //-->
- <Class name="~org\.apache\.bookkeeper\.proto\.BookkeeperProtocol.*" />
+ <Class
name="~org\.apache\.bookkeeper\.proto\.(LedgerMetadataFormat|LedgerRereplicationLayoutFormat|UnderreplicatedLedgerFormat|CookieFormat|LockDataFormat|AuditorVoteFormat|CheckAllLedgersFormat|PlacementPolicyCheckFormat|ReplicasCheckFormat|BookieServiceInfoFormat|LightProtoCodec)(\$.*)?"
/>
</Match>
+ <!-- lightproto-generated classes from DbLedgerStorageDataFormats.proto -->
<Match>
- <!-- generated code, we can't be held responsible for findbugs in it //-->
- <Class
name="~org\.apache\.bookkeeper\.bookie\.storage\.ldb\.DbLedgerStorageDataFormats.*"
/>
+ <Class
name="~org\.apache\.bookkeeper\.bookie\.storage\.ldb\.(LedgerData|LightProtoCodec)(\$.*)?"
/>
</Match>
<Match>
<!-- generated code, we can't be held responsible for findbugs in it //-->
diff --git
a/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManagerBenchmark.java
b/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManagerBenchmark.java
index cdf8f5a1ed..5a7fede926 100644
---
a/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManagerBenchmark.java
+++
b/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManagerBenchmark.java
@@ -25,7 +25,7 @@ import io.netty.buffer.PooledByteBufAllocator;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
-import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.DigestType;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat.DigestType;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
diff --git
a/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java
b/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java
index 7201aba8cf..39cb2b6069 100644
---
a/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java
+++
b/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java
@@ -27,7 +27,7 @@ import io.netty.buffer.Unpooled;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
-import org.apache.bookkeeper.proto.DataFormats.LedgerMetadataFormat.DigestType;
+import org.apache.bookkeeper.proto.LedgerMetadataFormat.DigestType;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
diff --git a/pom.xml b/pom.xml
index c7ae8ee878..3a9f2a9af1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -220,6 +220,7 @@
<nar-maven-plugin.version>3.10.1</nar-maven-plugin.version>
<os-maven-plugin.version>1.7.1</os-maven-plugin.version>
<protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>
+ <lightproto-maven-plugin.version>0.7.0</lightproto-maven-plugin.version>
<puppycrawl.checkstyle.version>9.3</puppycrawl.checkstyle.version>
<spotbugs-maven-plugin.version>4.7.3.2</spotbugs-maven-plugin.version>
<build-helper-maven-plugin.version>3.6.0</build-helper-maven-plugin.version>
