This is an automated email from the ASF dual-hosted git repository.
lhotari 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 c2866286ce Migrate stream non-gRPC protos to LightProto (#4781)
c2866286ce is described below
commit c2866286cebc2c945ce86e741fc50fbb46a58077
Author: Matteo Merli <[email protected]>
AuthorDate: Thu May 7 22:55:35 2026 -0700
Migrate stream non-gRPC protos to LightProto (#4781)
* 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 stream non-gRPC protos to LightProto
Migrates the stream module's non-gRPC proto definitions to LightProto:
- stream/statelib/src/main/proto/kv.proto (KV state-store commands)
- stream/proto/src/main/proto/cluster.proto (cluster metadata/assignment)
Both files use proto3, including oneof (kv.proto) and map (cluster.proto),
which LightProto handles directly.
For stream/proto, cluster.proto is moved to a parallel proto-lightproto/
directory so the existing protobuf-maven-plugin keeps generating the rest
of the protos (which still depend on the gRPC service stubs). Java sources
that touched the migrated types are updated to use the LightProto
mutable-instance API instead of the protobuf-java builder pattern.
This change is independent of the other in-flight LightProto work in
bookkeeper-proto and bookkeeper-server. gRPC service migration is left
for a follow-up.
* Fix checkstyle: remove unused ServerAssignmentData import
---
stream/proto/pom.xml | 27 +++++++++++
.../main/{proto => proto-lightproto}/cluster.proto | 0
stream/statelib/pom.xml | 30 ++++++------
.../statelib/impl/kv/KVCommandProcessor.java | 15 +++---
.../bookkeeper/statelib/impl/kv/KVUtils.java | 52 ++++++---------------
.../statelib/impl/kv/RocksdbKVAsyncStore.java | 18 +++-----
.../impl/cluster/InMemClusterMetadataStore.java | 6 +--
.../impl/cluster/ZkClusterMetadataStore.java | 36 +++++++++------
.../impl/sc/DefaultStorageContainerController.java | 54 ++++++++++------------
.../storage/impl/sc/ZkStorageContainerManager.java | 37 +++++++++------
.../cluster/InMemClusterMetadataStoreTest.java | 48 +++++++------------
.../impl/cluster/ZkClusterMetadataStoreTest.java | 40 ++++++----------
.../sc/DefaultStorageContainerControllerTest.java | 52 ++++++++++++---------
.../impl/sc/ZkStorageContainerManagerTest.java | 39 ++++------------
14 files changed, 214 insertions(+), 240 deletions(-)
diff --git a/stream/proto/pom.xml b/stream/proto/pom.xml
index 108d69aca6..d4e0fe6844 100644
--- a/stream/proto/pom.xml
+++ b/stream/proto/pom.xml
@@ -69,6 +69,15 @@
</extension>
</extensions>
<plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes combine.children="append">
+ <exclude>target/generated-sources/lightproto/**</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
@@ -96,6 +105,24 @@
</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-lightproto/cluster.proto</source>
+ </sources>
+
<targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
diff --git a/stream/proto/src/main/proto/cluster.proto
b/stream/proto/src/main/proto-lightproto/cluster.proto
similarity index 100%
rename from stream/proto/src/main/proto/cluster.proto
rename to stream/proto/src/main/proto-lightproto/cluster.proto
diff --git a/stream/statelib/pom.xml b/stream/statelib/pom.xml
index 73c1b349c1..804c22bad5 100644
--- a/stream/statelib/pom.xml
+++ b/stream/statelib/pom.xml
@@ -61,8 +61,8 @@
<artifactId>rocksdbjni</artifactId>
</dependency>
<dependency>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-buffer</artifactId>
</dependency>
<dependency>
<groupId>org.apache.distributedlog</groupId>
@@ -104,25 +104,27 @@
</dependency>
</dependencies>
<build>
- <extensions>
- <extension>
- <groupId>kr.motd.maven</groupId>
- <artifactId>os-maven-plugin</artifactId>
- <version>${os-maven-plugin.version}</version>
- </extension>
- </extensions>
<plugins>
<plugin>
- <groupId>org.xolstice.maven.plugins</groupId>
- <artifactId>protobuf-maven-plugin</artifactId>
- <version>${protobuf-maven-plugin.version}</version>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
<configuration>
-
<protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
+ <excludes combine.children="append">
+ <exclude>target/generated-sources/lightproto/**</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>io.streamnative.lightproto</groupId>
+ <artifactId>lightproto-maven-plugin</artifactId>
+ <version>${lightproto-maven-plugin.version}</version>
+ <configuration>
+
<targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
</configuration>
<executions>
<execution>
<goals>
- <goal>compile</goal>
+ <goal>generate</goal>
</goals>
</execution>
</executions>
diff --git
a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVCommandProcessor.java
b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVCommandProcessor.java
index 1eb9009a4c..d535106904 100644
---
a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVCommandProcessor.java
+++
b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVCommandProcessor.java
@@ -20,7 +20,6 @@ package org.apache.bookkeeper.statelib.impl.kv;
import static org.apache.bookkeeper.statelib.impl.kv.KVUtils.newCommand;
-import com.google.protobuf.InvalidProtocolBufferException;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import lombok.AccessLevel;
@@ -48,8 +47,8 @@ class KVCommandProcessor implements
CommandProcessor<RocksdbKVStore<byte[], byte
Command command;
try {
command = newCommand(cmdBuf);
- } catch (InvalidProtocolBufferException e) {
- log.error().attr("buffer", cmdBuf).attr("txid", revision)
+ } catch (RuntimeException e) {
+ log.error().exception(e).attr("buffer", cmdBuf).attr("txid",
revision)
.log("Invalid kv command found");
// TODO: better to handle this
return;
@@ -74,8 +73,8 @@ class KVCommandProcessor implements
CommandProcessor<RocksdbKVStore<byte[], byte
private void applyPutCommand(long revision, Command command,
RocksdbKVStore<byte[], byte[]> store) {
PutRequest putReq = command.getPutReq();
- byte[] keyBytes = putReq.getKey().toByteArray();
- byte[] valBytes = putReq.getValue().toByteArray();
+ byte[] keyBytes = putReq.getKey();
+ byte[] valBytes = putReq.getValue();
@Cleanup("release") ByteBuf serializedValBuf =
KVUtils.serialize(valBytes, revision);
byte[] serializedValBytes = ByteBufUtil.getBytes(serializedValBuf);
store.put(keyBytes, serializedValBytes, revision);
@@ -83,8 +82,8 @@ class KVCommandProcessor implements
CommandProcessor<RocksdbKVStore<byte[], byte
private void applyPutIfAbsentCommand(long revision, Command command,
RocksdbKVStore<byte[], byte[]> store) {
PutRequest putReq = command.getPutReq();
- byte[] keyBytes = putReq.getKey().toByteArray();
- byte[] valBytes = putReq.getValue().toByteArray();
+ byte[] keyBytes = putReq.getKey();
+ byte[] valBytes = putReq.getValue();
@Cleanup("release") ByteBuf serializedValBuf =
KVUtils.serialize(valBytes, revision);
byte[] serializedValBytes = ByteBufUtil.getBytes(serializedValBuf);
store.putIfAbsent(keyBytes, serializedValBytes, revision);
@@ -92,7 +91,7 @@ class KVCommandProcessor implements
CommandProcessor<RocksdbKVStore<byte[], byte
private void applyDeleteCommand(long revision, Command command,
RocksdbKVStore<byte[], byte[]> store) {
DeleteRequest delReq = command.getDelReq();
- byte[] keyBytes = delReq.getKey().toByteArray();
+ byte[] keyBytes = delReq.getKey();
store.delete(keyBytes, revision);
}
}
diff --git
a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVUtils.java
b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVUtils.java
index 337b403e13..12d8e52e3e 100644
---
a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVUtils.java
+++
b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVUtils.java
@@ -18,22 +18,14 @@
*/
package org.apache.bookkeeper.statelib.impl.kv;
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.google.protobuf.UnsafeByteOperations;
import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufOutputStream;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.util.ReferenceCountUtil;
-import java.io.IOException;
import lombok.AccessLevel;
import lombok.CustomLog;
import lombok.NoArgsConstructor;
import org.apache.bookkeeper.common.coder.Coder;
import org.apache.bookkeeper.proto.statestore.kv.Command;
-import org.apache.bookkeeper.proto.statestore.kv.DeleteRequest;
-import org.apache.bookkeeper.proto.statestore.kv.NopRequest;
-import org.apache.bookkeeper.proto.statestore.kv.PutIfAbsentRequest;
-import org.apache.bookkeeper.proto.statestore.kv.PutRequest;
/**
* Utils for kv stores.
@@ -42,9 +34,13 @@ import org.apache.bookkeeper.proto.statestore.kv.PutRequest;
@NoArgsConstructor(access = AccessLevel.PRIVATE)
final class KVUtils {
- static final Command NOP_CMD = Command.newBuilder()
- .setNopReq(NopRequest.newBuilder().build())
- .build();
+ static final Command NOP_CMD = newNopCommand();
+
+ private static Command newNopCommand() {
+ Command cmd = new Command();
+ cmd.setNopReq();
+ return cmd;
+ }
static ByteBuf serialize(ByteBuf valBuf, long revision) {
int serializedSize = valBuf.readableBytes() + Long.BYTES;
@@ -68,41 +64,21 @@ final class KVUtils {
return valCoder.decode(valBuf);
}
- static Command newCommand(ByteBuf cmdBuf) throws
InvalidProtocolBufferException {
- return Command.parseFrom(cmdBuf.nioBuffer());
+ static Command newCommand(ByteBuf cmdBuf) {
+ Command cmd = new Command();
+ cmd.parseFrom(cmdBuf, cmdBuf.readableBytes());
+ return cmd;
}
- static ByteBuf newCommandBuf(Command cmd) throws IOException {
+ static ByteBuf newCommandBuf(Command cmd) {
ByteBuf buf =
PooledByteBufAllocator.DEFAULT.buffer(cmd.getSerializedSize());
try {
- cmd.writeTo(new ByteBufOutputStream(buf));
- } catch (IOException e) {
+ cmd.writeTo(buf);
+ } catch (RuntimeException e) {
ReferenceCountUtil.release(buf);
throw e;
}
return buf;
}
- static PutRequest newPutRequest(byte[] keyBytes,
- byte[] valBytes) {
- return PutRequest.newBuilder()
- .setKey(UnsafeByteOperations.unsafeWrap(keyBytes))
- .setValue(UnsafeByteOperations.unsafeWrap(valBytes))
- .build();
- }
-
- static PutIfAbsentRequest newPutIfAbsentRequest(byte[] keyBytes,
- byte[] valBytes) {
- return PutIfAbsentRequest.newBuilder()
- .setKey(UnsafeByteOperations.unsafeWrap(keyBytes))
- .setValue(UnsafeByteOperations.unsafeWrap(valBytes))
- .build();
- }
-
- static DeleteRequest newDeleteRequest(byte[] keyBytes) {
- return DeleteRequest.newBuilder()
- .setKey(UnsafeByteOperations.unsafeWrap(keyBytes))
- .build();
- }
-
}
diff --git
a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVAsyncStore.java
b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVAsyncStore.java
index 63150d11d5..dc982a9e9b 100644
---
a/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVAsyncStore.java
+++
b/stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVAsyncStore.java
@@ -22,7 +22,6 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.util.ReferenceCountUtil;
-import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
import org.apache.bookkeeper.common.coder.Coder;
@@ -91,9 +90,8 @@ public class RocksdbKVAsyncStore<K, V>
byte[] keyBytes = keyCoder.encode(key);
byte[] valBytes = valCoder.encode(value);
- Command command = Command.newBuilder()
- .setPutReq(KVUtils.newPutRequest(keyBytes, valBytes))
- .build();
+ Command command = new Command();
+ command.setPutReq().setKey(keyBytes).setValue(valBytes);
return writeCommandReturnTxId(command).thenApplyAsync((revision) -> {
ByteBuf serializedBuf = KVUtils.serialize(valBytes, revision);
try {
@@ -112,9 +110,8 @@ public class RocksdbKVAsyncStore<K, V>
byte[] keyBytes = keyCoder.encode(key);
byte[] valBytes = valCoder.encode(value);
- Command command = Command.newBuilder()
- .setPutIfAbsentReq(KVUtils.newPutIfAbsentRequest(keyBytes,
valBytes))
- .build();
+ Command command = new Command();
+ command.setPutIfAbsentReq().setKey(keyBytes).setValue(valBytes);
return writeCommandReturnTxId(command).thenApplyAsync((revision) -> {
ByteBuf serializedBuf = KVUtils.serialize(valBytes, revision);
try {
@@ -135,9 +132,8 @@ public class RocksdbKVAsyncStore<K, V>
public CompletableFuture<V> delete(K key) {
byte[] keyBytes = keyCoder.encode(key);
- Command command = Command.newBuilder()
- .setDelReq(KVUtils.newDeleteRequest(keyBytes))
- .build();
+ Command command = new Command();
+ command.setDelReq().setKey(keyBytes);
return writeCommandReturnTxId(command).thenApplyAsync((revision) -> {
byte[] prevValue = localStore.delete(keyBytes, revision);
if (null == prevValue) {
@@ -171,7 +167,7 @@ public class RocksdbKVAsyncStore<K, V>
try {
ByteBuf cmdBuf = KVUtils.newCommandBuf(command);
return writeCommandBufReturnTxId(cmdBuf);
- } catch (IOException e) {
+ } catch (RuntimeException e) {
return FutureUtils.exception(e);
}
}
diff --git
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStore.java
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStore.java
index bea4b23091..645ced4505 100644
---
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStore.java
+++
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStore.java
@@ -56,10 +56,8 @@ public class InMemClusterMetadataStore implements
ClusterMetadataStore {
@Override
public synchronized boolean initializeCluster(int numStorageContainers,
Optional<String>
segmentStorePath) {
- this.metadata = ClusterMetadata.newBuilder()
- .setNumStorageContainers(numStorageContainers)
- .build();
- this.assignmentData = ClusterAssignmentData.newBuilder().build();
+ this.metadata = new
ClusterMetadata().setNumStorageContainers(numStorageContainers);
+ this.assignmentData = new ClusterAssignmentData();
return true;
}
diff --git
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStore.java
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStore.java
index e583fc2959..4f9f8f53c0 100644
---
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStore.java
+++
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStore.java
@@ -25,7 +25,6 @@ import static
org.apache.bookkeeper.stream.storage.StorageConstants.getServersPa
import static
org.apache.bookkeeper.stream.storage.StorageConstants.getStoragePath;
import static
org.apache.bookkeeper.stream.storage.StorageConstants.getWritableServersPath;
-import com.google.protobuf.InvalidProtocolBufferException;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
@@ -95,11 +94,8 @@ public class ZkClusterMetadataStore implements
ClusterMetadataStore {
@Override
public boolean initializeCluster(int numStorageContainers,
Optional<String> segmentStorePath) {
- ClusterMetadata metadata = ClusterMetadata.newBuilder()
- .setNumStorageContainers(numStorageContainers)
- .build();
- ClusterAssignmentData assignmentData =
ClusterAssignmentData.newBuilder()
- .build();
+ ClusterMetadata metadata = new
ClusterMetadata().setNumStorageContainers(numStorageContainers);
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
try {
// we are using dlog for the storage backend, so we need to
initialize the dlog namespace
BKDLConfig dlogConfig = new BKDLConfig(
@@ -130,10 +126,16 @@ public class ZkClusterMetadataStore implements
ClusterMetadataStore {
public ClusterAssignmentData getClusterAssignmentData() {
try {
byte[] data = client.getData().forPath(zkClusterAssignmentPath);
- return ClusterAssignmentData.parseFrom(data);
- } catch (InvalidProtocolBufferException ie) {
- throw new StorageRuntimeException("The cluster assignment data
from zookeeper @"
- + zkClusterAssignmentPath + " is corrupted", ie);
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
+ try {
+ assignmentData.parseFrom(data);
+ } catch (RuntimeException ie) {
+ throw new StorageRuntimeException("The cluster assignment data
from zookeeper @"
+ + zkClusterAssignmentPath + " is corrupted", ie);
+ }
+ return assignmentData;
+ } catch (StorageRuntimeException e) {
+ throw e;
} catch (Exception e) {
throw new StorageRuntimeException("Failed to fetch cluster
assignment data from zookeeper @"
+ zkClusterAssignmentPath, e);
@@ -193,10 +195,16 @@ public class ZkClusterMetadataStore implements
ClusterMetadataStore {
public ClusterMetadata getClusterMetadata() {
try {
byte[] data = client.getData().forPath(zkClusterMetadataPath);
- return ClusterMetadata.parseFrom(data);
- } catch (InvalidProtocolBufferException ie) {
- throw new StorageRuntimeException("The cluster metadata from
zookeeper @"
- + zkClusterMetadataPath + " is corrupted", ie);
+ ClusterMetadata metadata = new ClusterMetadata();
+ try {
+ metadata.parseFrom(data);
+ } catch (RuntimeException ie) {
+ throw new StorageRuntimeException("The cluster metadata from
zookeeper @"
+ + zkClusterMetadataPath + " is corrupted", ie);
+ }
+ return metadata;
+ } catch (StorageRuntimeException e) {
+ throw e;
} catch (Exception e) {
throw new StorageRuntimeException("Failed to fetch cluster
metadata from zookeeper @"
+ zkClusterMetadataPath, e);
diff --git
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerController.java
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerController.java
index ecafa8fcf4..327499e065 100644
---
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerController.java
+++
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerController.java
@@ -81,14 +81,15 @@ public class DefaultStorageContainerController implements
StorageContainerContro
// 1. get current server assignments
Map<BookieId, Set<Long>> currentServerAssignments;
try {
- currentServerAssignments = currentState.getServersMap()
- .entrySet()
- .stream()
- .collect(Collectors.toMap(e1 -> {
- return BookieId.parse(e1.getKey());
- },
- e2 ->
e2.getValue().getContainersList().stream().collect(Collectors.toSet())
- ));
+ currentServerAssignments = Maps.newHashMap();
+ Map<BookieId, Set<Long>> finalAssignments =
currentServerAssignments;
+ currentState.forEachServers((server, assignment) -> {
+ Set<Long> containers = Sets.newHashSet();
+ for (int i = 0; i < assignment.getContainersCount(); i++) {
+ containers.add(assignment.getContainerAt(i));
+ }
+ finalAssignments.put(BookieId.parse(server), containers);
+ });
} catch (UncheckedExecutionException uee) {
log.warn()
.attr("currentState", currentState)
@@ -181,15 +182,14 @@ public class DefaultStorageContainerController implements
StorageContainerContro
}
// 9. the new ideal state is computed, finalize it
- Map<String, ServerAssignmentData> newAssignmentMap = Maps.newHashMap();
- assignmentQueue.forEach(assignment -> newAssignmentMap.put(
- assignment.getKey().toString(),
- ServerAssignmentData.newBuilder()
- .addAllContainers(assignment.getValue())
- .build()));
- return ClusterAssignmentData.newBuilder()
- .putAllServers(newAssignmentMap)
- .build();
+ ClusterAssignmentData result = new ClusterAssignmentData();
+ for (Pair<BookieId, LinkedList<Long>> assignment : assignmentQueue) {
+ ServerAssignmentData server =
result.putServers(assignment.getKey().toString());
+ for (long container : assignment.getValue()) {
+ server.addContainer(container);
+ }
+ }
+ return result;
}
static ClusterAssignmentData initializeIdealState(ClusterMetadata
clusterMetadata,
@@ -202,24 +202,20 @@ public class DefaultStorageContainerController implements
StorageContainerContro
int numTotalContainers = (int)
clusterMetadata.getNumStorageContainers();
int numContainersPerServer = numTotalContainers /
currentCluster.size();
- Map<String, ServerAssignmentData> assignmentMap = Maps.newHashMap();
+ ClusterAssignmentData result = new ClusterAssignmentData();
for (int serverIdx = 0; serverIdx < serverList.size(); serverIdx++) {
BookieId server = serverList.get(serverIdx);
+ ServerAssignmentData assignmentData =
result.putServers(server.toString());
int finalServerIdx = serverIdx;
- ServerAssignmentData assignmentData =
ServerAssignmentData.newBuilder()
- .addAllContainers(
- LongStream.rangeClosed(0, numContainersPerServer).boxed()
- .map(j -> j * numServers + finalServerIdx)
- .filter(containerId -> containerId <
numTotalContainers)
- .collect(Collectors.toSet()))
- .build();
- assignmentMap.put(server.toString(), assignmentData);
+ LongStream.rangeClosed(0, numContainersPerServer)
+ .map(j -> j * numServers + finalServerIdx)
+ .filter(containerId -> containerId < numTotalContainers)
+ .distinct()
+ .forEach(assignmentData::addContainer);
}
- return ClusterAssignmentData.newBuilder()
- .putAllServers(assignmentMap)
- .build();
+ return result;
}
}
diff --git
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManager.java
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManager.java
index 6868103a58..02d1d69c1d 100644
---
a/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManager.java
+++
b/stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManager.java
@@ -25,6 +25,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -33,7 +34,6 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
-import java.util.stream.Collectors;
import lombok.AccessLevel;
import lombok.CustomLog;
import lombok.Getter;
@@ -59,6 +59,8 @@ public class ZkStorageContainerManager
extends AbstractLifecycleComponent<StorageConfiguration>
implements StorageContainerManager, Consumer<Void> {
+ private static final ServerAssignmentData EMPTY_SERVER_ASSIGNMENT = new
ServerAssignmentData();
+
private final Endpoint endpoint;
private final ClusterMetadataStore metadataStore;
private final StorageContainerRegistry registry;
@@ -161,11 +163,9 @@ public class ZkStorageContainerManager
return false;
}
- Map<Endpoint, ServerAssignmentData> newAssignmentMap =
clusterAssignmentData.getServersMap().entrySet()
- .stream()
- .collect(Collectors.toMap(
- e -> NetUtils.parseEndpoint(e.getKey()),
- e -> e.getValue()));
+ Map<Endpoint, ServerAssignmentData> newAssignmentMap = new HashMap<>();
+ clusterAssignmentData.forEachServers(
+ (server, assignment) ->
newAssignmentMap.put(NetUtils.parseEndpoint(server), assignment));
Set<Endpoint> oldAssignedServers = clusterAssignmentMap.keySet();
Set<Endpoint> newAssignedServers = newAssignmentMap.keySet();
@@ -191,7 +191,9 @@ public class ZkStorageContainerManager
serversJoined.forEach(ep -> {
ServerAssignmentData sad = newAssignmentMap.get(ep);
if (null != sad) {
- sad.getContainersList().forEach(container ->
containerAssignmentMap.put(container, ep));
+ for (int i = 0; i < sad.getContainersCount(); i++) {
+ containerAssignmentMap.put(sad.getContainerAt(i), ep);
+ }
}
});
}
@@ -204,7 +206,9 @@ public class ZkStorageContainerManager
serversLeft.forEach(ep -> {
ServerAssignmentData sad = oldAssignmentMap.get(ep);
if (null != sad) {
- sad.getContainersList().forEach(container ->
containerAssignmentMap.remove(container, ep));
+ for (int i = 0; i < sad.getContainersCount(); i++) {
+ containerAssignmentMap.remove(sad.getContainerAt(i), ep);
+ }
}
});
}
@@ -214,8 +218,8 @@ public class ZkStorageContainerManager
Map<Endpoint,
ServerAssignmentData> newAssignmentMap) {
commonServers.forEach(ep -> {
- ServerAssignmentData oldSad = oldAssignmentMap.getOrDefault(ep,
ServerAssignmentData.getDefaultInstance());
- ServerAssignmentData newSad = newAssignmentMap.getOrDefault(ep,
ServerAssignmentData.getDefaultInstance());
+ ServerAssignmentData oldSad = oldAssignmentMap.getOrDefault(ep,
EMPTY_SERVER_ASSIGNMENT);
+ ServerAssignmentData newSad = newAssignmentMap.getOrDefault(ep,
EMPTY_SERVER_ASSIGNMENT);
if (oldSad.equals(newSad)) {
return;
@@ -225,8 +229,12 @@ public class ZkStorageContainerManager
.attr("oldAssignment", oldSad)
.attr("newAssignment", newSad)
.log("Server assignment is changed");
- oldSad.getContainersList().forEach(container ->
containerAssignmentMap.remove(container, ep));
- newSad.getContainersList().forEach(container ->
containerAssignmentMap.put(container, ep));
+ for (int i = 0; i < oldSad.getContainersCount(); i++) {
+ containerAssignmentMap.remove(oldSad.getContainerAt(i),
ep);
+ }
+ for (int i = 0; i < newSad.getContainersCount(); i++) {
+ containerAssignmentMap.put(newSad.getContainerAt(i), ep);
+ }
}
});
@@ -239,7 +247,10 @@ public class ZkStorageContainerManager
}
private void processMyAssignment(ServerAssignmentData myAssignment) {
- Set<Long> assignedContainerSet =
myAssignment.getContainersList().stream().collect(Collectors.toSet());
+ Set<Long> assignedContainerSet = Sets.newHashSet();
+ for (int i = 0; i < myAssignment.getContainersCount(); i++) {
+ assignedContainerSet.add(myAssignment.getContainerAt(i));
+ }
Set<Long> liveContainerSet = Sets.newHashSet(liveContainers.keySet());
Set<Long> containersToStart =
diff --git
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStoreTest.java
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStoreTest.java
index 447f934822..cba8a53c00 100644
---
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStoreTest.java
+++
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStoreTest.java
@@ -59,10 +59,10 @@ public class InMemClusterMetadataStoreTest {
@Test
public void testUninitialized() {
assertEquals(
-
ClusterMetadata.newBuilder().setNumStorageContainers(NUM_STORAGE_CONTAINERS).build(),
+ new
ClusterMetadata().setNumStorageContainers(NUM_STORAGE_CONTAINERS),
store.getClusterMetadata());
assertEquals(
- ClusterAssignmentData.newBuilder().build(),
+ new ClusterAssignmentData(),
store.getClusterAssignmentData());
}
@@ -71,47 +71,37 @@ public class InMemClusterMetadataStoreTest {
int numStorageContainers = 2048;
store.initializeCluster(numStorageContainers);
assertEquals(
-
ClusterMetadata.newBuilder().setNumStorageContainers(numStorageContainers).build(),
+ new
ClusterMetadata().setNumStorageContainers(numStorageContainers),
store.getClusterMetadata());
assertEquals(
- ClusterAssignmentData.newBuilder().build(),
+ new ClusterAssignmentData(),
store.getClusterAssignmentData());
}
@Test
public void testUpdateClusterMetadata() {
int numStorageContainers = 4096;
- ClusterMetadata metadata = ClusterMetadata.newBuilder()
- .setNumStorageContainers(numStorageContainers)
- .build();
+ ClusterMetadata metadata = new
ClusterMetadata().setNumStorageContainers(numStorageContainers);
store.updateClusterMetadata(metadata);
assertEquals(metadata, store.getClusterMetadata());
}
@Test
public void testUpdateClusterAssignmentData() {
- ClusterAssignmentData assignmentData =
ClusterAssignmentData.newBuilder()
- .putServers(
- "server-0",
- ServerAssignmentData.newBuilder()
- .addContainers(1L)
- .addContainers(2L)
- .build())
- .build();
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
+ ServerAssignmentData server0 = assignmentData.putServers("server-0");
+ server0.addContainer(1L);
+ server0.addContainer(2L);
store.updateClusterAssignmentData(assignmentData);
assertEquals(assignmentData, store.getClusterAssignmentData());
}
@Test
public void testWatchClusterAssignmentData() {
- ClusterAssignmentData assignmentData =
ClusterAssignmentData.newBuilder()
- .putServers(
- "server-0",
- ServerAssignmentData.newBuilder()
- .addContainers(1L)
- .addContainers(2L)
- .build())
- .build();
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
+ ServerAssignmentData server0 = assignmentData.putServers("server-0");
+ server0.addContainer(1L);
+ server0.addContainer(2L);
@Cleanup("shutdown")
ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -129,14 +119,10 @@ public class InMemClusterMetadataStoreTest {
@Test
public void testUnwatchClusterAssignmentData() throws Exception {
- ClusterAssignmentData assignmentData =
ClusterAssignmentData.newBuilder()
- .putServers(
- "server-0",
- ServerAssignmentData.newBuilder()
- .addContainers(1L)
- .addContainers(2L)
- .build())
- .build();
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
+ ServerAssignmentData server0 = assignmentData.putServers("server-0");
+ server0.addContainer(1L);
+ server0.addContainer(2L);
@Cleanup("shutdown")
ExecutorService executor = Executors.newSingleThreadExecutor();
diff --git
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStoreTest.java
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStoreTest.java
index bcf34331e9..4e87451d61 100644
---
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStoreTest.java
+++
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStoreTest.java
@@ -114,37 +114,27 @@ public class ZkClusterMetadataStoreTest extends
ZooKeeperClusterTestCase {
@Test
public void testUpdateClusterMetadata() {
int numStorageContainers = 4096;
- ClusterMetadata metadata = ClusterMetadata.newBuilder()
- .setNumStorageContainers(numStorageContainers)
- .build();
+ ClusterMetadata metadata = new
ClusterMetadata().setNumStorageContainers(numStorageContainers);
store.updateClusterMetadata(metadata);
assertEquals(metadata, store.getClusterMetadata());
}
@Test
public void testUpdateClusterAssignmentData() {
- ClusterAssignmentData assignmentData =
ClusterAssignmentData.newBuilder()
- .putServers(
- "server-0",
- ServerAssignmentData.newBuilder()
- .addContainers(1L)
- .addContainers(2L)
- .build())
- .build();
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
+ ServerAssignmentData server0 = assignmentData.putServers("server-0");
+ server0.addContainer(1L);
+ server0.addContainer(2L);
store.updateClusterAssignmentData(assignmentData);
assertEquals(assignmentData, store.getClusterAssignmentData());
}
@Test
public void testWatchClusterAssignmentData() {
- ClusterAssignmentData assignmentData =
ClusterAssignmentData.newBuilder()
- .putServers(
- "server-0",
- ServerAssignmentData.newBuilder()
- .addContainers(1L)
- .addContainers(2L)
- .build())
- .build();
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
+ ServerAssignmentData server0 = assignmentData.putServers("server-0");
+ server0.addContainer(1L);
+ server0.addContainer(2L);
@Cleanup("shutdown")
ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -162,14 +152,10 @@ public class ZkClusterMetadataStoreTest extends
ZooKeeperClusterTestCase {
@Test
public void testUnwatchClusterAssignmentData() throws Exception {
- ClusterAssignmentData assignmentData =
ClusterAssignmentData.newBuilder()
- .putServers(
- "server-0",
- ServerAssignmentData.newBuilder()
- .addContainers(1L)
- .addContainers(2L)
- .build())
- .build();
+ ClusterAssignmentData assignmentData = new ClusterAssignmentData();
+ ServerAssignmentData server0 = assignmentData.putServers("server-0");
+ server0.addContainer(1L);
+ server0.addContainer(2L);
@Cleanup("shutdown")
ExecutorService executor = Executors.newSingleThreadExecutor();
diff --git
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerControllerTest.java
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerControllerTest.java
index f6befb0487..009bbd9926 100644
---
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerControllerTest.java
+++
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/DefaultStorageContainerControllerTest.java
@@ -57,16 +57,26 @@ public class DefaultStorageContainerControllerTest {
public DefaultStorageContainerControllerTest() {
this.controller = new DefaultStorageContainerController();
- this.clusterMetadata = ClusterMetadata.newBuilder()
- .setNumStorageContainers(NUM_STORAGE_CONTAINERS)
- .build();
- this.currentAssignment = ClusterAssignmentData.newBuilder()
- .putServers("default-server", ServerAssignmentData.newBuilder()
- .addContainers(0L)
- .addContainers(1L)
- .addContainers(3L)
- .build())
- .build();
+ this.clusterMetadata = new
ClusterMetadata().setNumStorageContainers(NUM_STORAGE_CONTAINERS);
+ this.currentAssignment = new ClusterAssignmentData();
+ ServerAssignmentData defaultServer =
currentAssignment.putServers("default-server");
+ defaultServer.addContainer(0L);
+ defaultServer.addContainer(1L);
+ defaultServer.addContainer(3L);
+ }
+
+ private static List<Long> containersOf(ServerAssignmentData serverData) {
+ List<Long> containers =
Lists.newArrayListWithCapacity(serverData.getContainersCount());
+ for (int i = 0; i < serverData.getContainersCount(); i++) {
+ containers.add(serverData.getContainerAt(i));
+ }
+ return containers;
+ }
+
+ private static Map<String, ServerAssignmentData>
serversMap(ClusterAssignmentData assignment) {
+ Map<String, ServerAssignmentData> map = new
java.util.LinkedHashMap<>();
+ assignment.forEachServers(map::put);
+ return map;
}
@Test
@@ -129,7 +139,7 @@ public class DefaultStorageContainerControllerTest {
int numContainersPerServer = NUM_STORAGE_CONTAINERS / numServers;
int serverIdx = 0;
- for (Map.Entry<String, ServerAssignmentData> entry :
newAssignment.getServersMap().entrySet()) {
+ for (Map.Entry<String, ServerAssignmentData> entry :
serversMap(newAssignment).entrySet()) {
log.info().attr("server", entry.getKey()).attr("assignment",
entry.getValue())
.log("Check assignment for server");
@@ -139,7 +149,7 @@ public class DefaultStorageContainerControllerTest {
ServerAssignmentData serverData = entry.getValue();
assertEquals(numContainersPerServer,
serverData.getContainersCount());
- List<Long> containers =
Lists.newArrayList(serverData.getContainersList());
+ List<Long> containers = containersOf(serverData);
Collections.sort(containers);
assignedContainers.addAll(containers);
@@ -173,7 +183,7 @@ public class DefaultStorageContainerControllerTest {
int numEmptyServers = 0;
int numAssignedServers = 0;
int serverIdx = 0;
- for (Map.Entry<String, ServerAssignmentData> entry :
newAssignment.getServersMap().entrySet()) {
+ for (Map.Entry<String, ServerAssignmentData> entry :
serversMap(newAssignment).entrySet()) {
log.info().attr("server", entry.getKey()).attr("assignment",
entry.getValue())
.log("Check assignment for server");
@@ -188,7 +198,7 @@ public class DefaultStorageContainerControllerTest {
} else {
++numEmptyServers;
}
- List<Long> containers =
Lists.newArrayList(serverData.getContainersList());
+ List<Long> containers = containersOf(serverData);
Collections.sort(containers);
assignedContainers.addAll(containers);
@@ -209,7 +219,7 @@ public class DefaultStorageContainerControllerTest {
@Test
public void testComputeIdealStateFromEmptyAssignment() throws Exception {
- ClusterAssignmentData emptyAssignment =
ClusterAssignmentData.newBuilder().build();
+ ClusterAssignmentData emptyAssignment = new ClusterAssignmentData();
int numServers = 8;
Set<BookieId> currentCluster = newCluster(numServers);
@@ -224,7 +234,7 @@ public class DefaultStorageContainerControllerTest {
@Test
public void testComputeIdealStateIfClusterUnchanged() throws Exception {
- ClusterAssignmentData emptyAssignment =
ClusterAssignmentData.newBuilder().build();
+ ClusterAssignmentData emptyAssignment = new ClusterAssignmentData();
int numServers = 8;
Set<BookieId> currentCluster = newCluster(numServers);
@@ -245,7 +255,7 @@ public class DefaultStorageContainerControllerTest {
@Test
public void testComputeIdealStateWhenHostsRemoved() throws Exception {
- ClusterAssignmentData emptyAssignment =
ClusterAssignmentData.newBuilder().build();
+ ClusterAssignmentData emptyAssignment = new ClusterAssignmentData();
int numServers = 8;
Set<BookieId> currentCluster = newCluster(numServers);
@@ -268,7 +278,7 @@ public class DefaultStorageContainerControllerTest {
@Test
public void testComputeIdealStateWhenHostsAdded() throws Exception {
- ClusterAssignmentData emptyAssignment =
ClusterAssignmentData.newBuilder().build();
+ ClusterAssignmentData emptyAssignment = new ClusterAssignmentData();
int numServers = 4;
Set<BookieId> currentCluster = newCluster(numServers);
@@ -291,7 +301,7 @@ public class DefaultStorageContainerControllerTest {
@Test
public void testComputeIdealStateWhenHostsRemovedAdded() throws Exception {
- ClusterAssignmentData emptyAssignment =
ClusterAssignmentData.newBuilder().build();
+ ClusterAssignmentData emptyAssignment = new ClusterAssignmentData();
int numServers = 4;
Set<BookieId> currentCluster = newCluster(numServers);
@@ -318,7 +328,7 @@ public class DefaultStorageContainerControllerTest {
@Test
public void testComputeIdealStateWhenHasMoreServers() throws Exception {
- ClusterAssignmentData emptyAssignment =
ClusterAssignmentData.newBuilder().build();
+ ClusterAssignmentData emptyAssignment = new ClusterAssignmentData();
int numServers = 2 * NUM_STORAGE_CONTAINERS;
Set<BookieId> currentCluster = newCluster(numServers);
@@ -332,7 +342,7 @@ public class DefaultStorageContainerControllerTest {
@Test
public void testComputeIdealStateWhenScaleToMoreServers() throws Exception
{
- ClusterAssignmentData emptyAssignment =
ClusterAssignmentData.newBuilder().build();
+ ClusterAssignmentData emptyAssignment = new ClusterAssignmentData();
int numServers = 4;
Set<BookieId> currentCluster = newCluster(numServers);
diff --git
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManagerTest.java
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManagerTest.java
index bd33392a57..69cdf76257 100644
---
a/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManagerTest.java
+++
b/stream/storage/impl/src/test/java/org/apache/bookkeeper/stream/storage/impl/sc/ZkStorageContainerManagerTest.java
@@ -37,7 +37,6 @@ import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.common.testing.MoreAsserts;
import org.apache.bookkeeper.stats.NullStatsLogger;
import org.apache.bookkeeper.stream.proto.cluster.ClusterAssignmentData;
-import org.apache.bookkeeper.stream.proto.cluster.ServerAssignmentData;
import org.apache.bookkeeper.stream.proto.common.Endpoint;
import org.apache.bookkeeper.stream.storage.api.sc.StorageContainer;
import org.apache.bookkeeper.stream.storage.api.sc.StorageContainerFactory;
@@ -150,13 +149,8 @@ public class ZkStorageContainerManagerTest extends
ZooKeeperClusterTestCase {
.thenReturn(mockSc2);
// update assignment map
- ClusterAssignmentData cad = ClusterAssignmentData.newBuilder()
- .putServers(
- NetUtils.endpointToString(myEndpoint),
- ServerAssignmentData.newBuilder()
- .addContainers(containerId)
- .build())
- .build();
+ ClusterAssignmentData cad = new ClusterAssignmentData();
+
cad.putServers(NetUtils.endpointToString(myEndpoint)).addContainer(containerId);
clusterMetadataStore.updateClusterAssignmentData(cad);
// notify the container to complete startup
@@ -170,13 +164,8 @@ public class ZkStorageContainerManagerTest extends
ZooKeeperClusterTestCase {
// update assignment map to remove containerId and add containerId2
- ClusterAssignmentData newCad = ClusterAssignmentData.newBuilder()
- .putServers(
- NetUtils.endpointToString(myEndpoint),
- ServerAssignmentData.newBuilder()
- .addContainers(22L)
- .build())
- .build();
+ ClusterAssignmentData newCad = new ClusterAssignmentData();
+
newCad.putServers(NetUtils.endpointToString(myEndpoint)).addContainer(22L);
clusterMetadataStore.updateClusterAssignmentData(newCad);
// notify the container1 to stop and container2 to start
@@ -210,13 +199,8 @@ public class ZkStorageContainerManagerTest extends
ZooKeeperClusterTestCase {
.thenReturn(mockSc);
// update assignment map
- ClusterAssignmentData cad = ClusterAssignmentData.newBuilder()
- .putServers(
- NetUtils.endpointToString(myEndpoint),
- ServerAssignmentData.newBuilder()
- .addContainers(containerId)
- .build())
- .build();
+ ClusterAssignmentData cad = new ClusterAssignmentData();
+
cad.putServers(NetUtils.endpointToString(myEndpoint)).addContainer(containerId);
clusterMetadataStore.updateClusterAssignmentData(cad);
// wait until container start is called
@@ -226,7 +210,7 @@ public class ZkStorageContainerManagerTest extends
ZooKeeperClusterTestCase {
assertTrue(scManager.getPendingStartStopContainers().contains(containerId));
// now shutting the manager down
- cad = ClusterAssignmentData.newBuilder().build();
+ cad = new ClusterAssignmentData();
clusterMetadataStore.updateClusterAssignmentData(cad);
// the container should not be stopped since it is pending starting.
@@ -282,13 +266,8 @@ public class ZkStorageContainerManagerTest extends
ZooKeeperClusterTestCase {
scManager.start();
// update assignment map
- ClusterAssignmentData cad = ClusterAssignmentData.newBuilder()
- .putServers(
- NetUtils.endpointToString(myEndpoint),
- ServerAssignmentData.newBuilder()
- .addContainers(containerId)
- .build())
- .build();
+ ClusterAssignmentData cad = new ClusterAssignmentData();
+
cad.putServers(NetUtils.endpointToString(myEndpoint)).addContainer(containerId);
clusterMetadataStore.updateClusterAssignmentData(cad);
// wait until container start is called and verify it is not started.
