Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SERIALIZATION] Serialize Borker info in Clusterinfo #1721

Merged
merged 15 commits into from
May 27, 2023
Merged

[SERIALIZATION] Serialize Borker info in Clusterinfo #1721

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
2701825
serialize Broker info in ClusterInfo.
chaohengstudent May 9, 2023
5facee3
add Broker.BrokerImpl, Broker.DataFolder.DataFolderImpl and Config.eq…
chaohengstudent May 10, 2023
c45830a
Merge branch 'main' into serialize-borker-in-clusterinfo
chaohengstudent May 10, 2023
17b88cd
apply toList
chaohengstudent May 10, 2023
4483350
Separate ClusterInfo proto file.
chaohengstudent May 11, 2023
4d7499d
Make config into record.
chaohengstudent May 12, 2023
6f4658f
Make Broker into record.
chaohengstudent May 12, 2023
bb076c6
Merge branch 'main' into serialize-borker-in-clusterinfo
chaohengstudent May 16, 2023
159e922
Merge branch 'main' into serialize-borker-in-clusterinfo-IN-Process
chaohengstudent May 24, 2023
85984f0
update with main branch.
chaohengstudent May 24, 2023
7f978f5
fix proto file format.
chaohengstudent May 24, 2023
7d2991a
fix typo. reconstruct the code.
chaohengstudent May 25, 2023
1488f06
add toTopicPartition and toOuterClass.
chaohengstudent May 25, 2023
9ac9736
Merge branch 'main' into serialize-borker-in-clusterinfo-IN-Process
chaohengstudent May 27, 2023
7d5382b
fix conflict.
chaohengstudent May 27, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
123 changes: 108 additions & 15 deletions common/src/main/java/org/astraea/common/ByteUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.astraea.common.admin.Broker;
import org.astraea.common.admin.ClusterInfo;
import org.astraea.common.admin.Config;
import org.astraea.common.admin.NodeInfo;
Expand Down Expand Up @@ -185,16 +186,62 @@ public static byte[] toBytes(BeanObject value) {
public static byte[] toBytes(ClusterInfo value) {
return ClusterInfoOuterClass.ClusterInfo.newBuilder()
.setClusterId(value.clusterId())
.addAllNodeInfo(
value.nodes().stream()
.addAllBroker(
value.brokers().stream()
.map(
nodeInfo ->
ClusterInfoOuterClass.ClusterInfo.NodeInfo.newBuilder()
.setId(nodeInfo.id())
.setHost(nodeInfo.host())
.setPort(nodeInfo.port())
broker ->
ClusterInfoOuterClass.ClusterInfo.Broker.newBuilder()
.setId(broker.id())
.setHost(broker.host())
.setPort(broker.port())
.setIsController(broker.isController())
.putAllConfig(broker.config().raw())
.addAllDatafolder(
broker.dataFolders().stream()
.map(
dataFolder ->
ClusterInfoOuterClass.ClusterInfo.Broker.Datafolder
.newBuilder()
.setPath(dataFolder.path())
.putAllPartitionSizes(
dataFolder.partitionSizes().entrySet().stream()
.collect(
Collectors.toMap(
entry -> entry.getKey().toString(),
Map.Entry::getValue)))
.putAllOrphanPartitionSizes(
dataFolder
.orphanPartitionSizes()
.entrySet()
.stream()
.collect(
Collectors.toMap(
entry -> entry.getKey().toString(),
Map.Entry::getValue)))
.build())
.toList())
.addAllTopicPartitions(
broker.topicPartitions().stream()
.map(
tp ->
ClusterInfoOuterClass.ClusterInfo.Broker.TopicPartition
.newBuilder()
.setPartition(tp.partition())
.setTopic(tp.topic())
.build())
.toList())
.addAllTopicPartitionLeaders(
broker.topicPartitionLeaders().stream()
.map(
tp ->
ClusterInfoOuterClass.ClusterInfo.Broker.TopicPartition
.newBuilder()
.setPartition(tp.partition())
.setTopic(tp.topic())
.build())
.toList())
.build())
.collect(Collectors.toList()))
.toList())
.addAllTopic(
value.topics().values().stream()
.map(
Expand All @@ -208,7 +255,7 @@ public static byte[] toBytes(ClusterInfo value) {
.map(TopicPartition::partition)
.collect(Collectors.toList()))
.build())
.collect(Collectors.toList()))
.toList())
.addAllReplica(
value.replicas().stream()
.map(
Expand All @@ -217,7 +264,7 @@ public static byte[] toBytes(ClusterInfo value) {
.setTopic(replica.topic())
.setPartition(replica.partition())
.setNodeInfo(
ClusterInfoOuterClass.ClusterInfo.NodeInfo.newBuilder()
ClusterInfoOuterClass.ClusterInfo.Replica.NodeInfo.newBuilder()
.setId(replica.nodeInfo().id())
.setHost(replica.nodeInfo().host())
.setPort(replica.nodeInfo().port())
Expand All @@ -231,7 +278,7 @@ public static byte[] toBytes(ClusterInfo value) {
.setIsPreferredLeader(replica.isPreferredLeader())
.setPath(replica.path())
.build())
.collect(Collectors.toList()))
.toList())
.build()
.toByteArray();
}
Expand Down Expand Up @@ -326,10 +373,56 @@ public static ClusterInfo readClusterInfo(byte[] bytes) {
var outerClusterInfo = ClusterInfoOuterClass.ClusterInfo.parseFrom(bytes);
return ClusterInfo.of(
outerClusterInfo.getClusterId(),
outerClusterInfo.getNodeInfoList().stream()
outerClusterInfo.getBrokerList().stream()
.map(
nodeInfo -> NodeInfo.of(nodeInfo.getId(), nodeInfo.getHost(), nodeInfo.getPort()))
.collect(Collectors.toList()),
broker -> {
var host = broker.getHost();
var port = broker.getPort();
var id = broker.getId();
var isController = broker.getIsController();
var config = Config.of(broker.getConfigMap());
var dataFolders =
broker.getDatafolderList().stream()
.map(
datafolder -> {
var path = datafolder.getPath();
var partitionSizes =
datafolder.getPartitionSizesMap().entrySet().stream()
.collect(
Collectors.toMap(
entry -> TopicPartition.of(entry.getKey()),
Map.Entry::getValue));
var orphanPartitionSizes =
datafolder.getOrphanPartitionSizesMap().entrySet().stream()
.collect(
Collectors.toMap(
entry -> TopicPartition.of(entry.getKey()),
Map.Entry::getValue));
return (Broker.DataFolder)
new Broker.DataFolder.DataFolderImpl(
path, partitionSizes, orphanPartitionSizes);
})
.toList();
var topicPartitions =
broker.getTopicPartitionsList().stream()
.map(tp -> TopicPartition.of(tp.getTopic(), tp.getPartition()))
.collect(Collectors.toSet());
var topicPartitionLeaders =
broker.getTopicPartitionLeadersList().stream()
.map(tp -> TopicPartition.of(tp.getTopic(), tp.getPartition()))
.collect(Collectors.toSet());
return (NodeInfo)
new Broker.BrokerImpl(
host,
port,
id,
isController,
config,
dataFolders,
topicPartitions,
topicPartitionLeaders);
})
.toList(),
outerClusterInfo.getTopicList().stream()
.map(
protoTopic ->
Expand Down Expand Up @@ -377,7 +470,7 @@ public Set<TopicPartition> topicPartitions() {
.isPreferredLeader(replica.getIsPreferredLeader())
.path(replica.getPath())
.build())
.collect(Collectors.toList()));
.toList());
} catch (InvalidProtocolBufferException ex) {
throw new SerializationException(ex);
}
Expand Down
87 changes: 28 additions & 59 deletions common/src/main/java/org/astraea/common/admin/Broker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,25 +61,9 @@ static Broker of(
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

return (DataFolder)
new DataFolder() {

@Override
public String path() {
return path;
}

@Override
public Map<TopicPartition, Long> partitionSizes() {
return partitionSizes;
}

@Override
public Map<TopicPartition, Long> orphanPartitionSizes() {
return orphanPartitionSizes;
}
};
new DataFolder.DataFolderImpl(path, partitionSizes, orphanPartitionSizes);
})
.collect(Collectors.toList());
.toList();
var topicPartitionLeaders =
topics.stream()
.flatMap(
Expand All @@ -88,47 +72,15 @@ public Map<TopicPartition, Long> orphanPartitionSizes() {
.filter(p -> p.leader() != null && p.leader().id() == nodeInfo.id())
.map(p -> TopicPartition.of(topic.name(), p.partition())))
.collect(Collectors.toUnmodifiableSet());
return new Broker() {
@Override
public String host() {
return nodeInfo.host();
}

@Override
public int port() {
return nodeInfo.port();
}

@Override
public int id() {
return nodeInfo.id();
}

@Override
public boolean isController() {
return isController;
}

@Override
public Config config() {
return config;
}

@Override
public List<DataFolder> dataFolders() {
return folders;
}

@Override
public Set<TopicPartition> topicPartitions() {
return partitionsFromTopicDesc;
}

@Override
public Set<TopicPartition> topicPartitionLeaders() {
return topicPartitionLeaders;
}
};
return new BrokerImpl(
nodeInfo.host(),
nodeInfo.port(),
nodeInfo.id(),
isController,
config,
folders,
partitionsFromTopicDesc,
topicPartitionLeaders);
}

boolean isController();
Expand Down Expand Up @@ -166,5 +118,22 @@ interface DataFolder {
* @return topic partition located by this node but not traced by cluster
*/
Map<TopicPartition, Long> orphanPartitionSizes();

record DataFolderImpl(
String path,
Map<TopicPartition, Long> partitionSizes,
Map<TopicPartition, Long> orphanPartitionSizes)
implements DataFolder {}
}

record BrokerImpl(
chaohengstudent marked this conversation as resolved.
Show resolved Hide resolved
String host,
int port,
int id,
boolean isController,
Config config,
List<DataFolder> dataFolders,
Set<TopicPartition> topicPartitions,
Set<TopicPartition> topicPartitionLeaders)
implements Broker {}
}
8 changes: 8 additions & 0 deletions common/src/main/java/org/astraea/common/admin/Config.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,14 @@ public Map<String, String> raw() {
public Optional<String> value(String key) {
return Optional.ofNullable(configs.get(key));
}

@Override
public boolean equals(Object obj) {
chaohengstudent marked this conversation as resolved.
Show resolved Hide resolved
if (this == obj) return true;
if (obj == null || getClass() != obj.getClass()) return false;
var objConfig = (Config) obj;
return raw().equals(objConfig.raw());
}
};
}

Expand Down
33 changes: 28 additions & 5 deletions common/src/main/proto/org/astraea/common/generated/ClusterInfo.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,25 +3,43 @@ syntax = "proto3";
package org.astraea.common.generated;

message ClusterInfo{
string clusterId = 1;

repeated NodeInfo nodeInfo = 2;
message NodeInfo {
message Broker {
chaohengstudent marked this conversation as resolved.
Show resolved Hide resolved
message Datafolder {
string path = 1;
map<string, int64> partitionSizes = 2;
map<string, int64> orphanPartitionSizes = 3;
}

message TopicPartition {
int32 partition = 1;
string topic = 2;
}

int32 id = 1;
string host = 2;
int32 port = 3;
bool isController = 4;
map<string, string> config = 5;
repeated Datafolder datafolder = 6;
repeated TopicPartition topicPartitions = 7;
repeated TopicPartition topicPartitionLeaders = 8;
}

repeated Topic topic = 3;
message Topic {
string name = 1;
map<string, string> config = 2;
bool internal = 3;
repeated int32 partition = 4;
}

repeated Replica replica = 4;
message Replica{
message NodeInfo {
int32 id = 1;
string host = 2;
int32 port = 3;
}

string topic = 1;
int32 partition = 2;
NodeInfo nodeInfo = 3;
Expand All @@ -34,4 +52,9 @@ message ClusterInfo{
bool isPreferredLeader = 10;
string path = 11;
}

string clusterId = 1;
repeated Broker broker = 2;
repeated Topic topic = 3;
repeated Replica replica = 4;
}
3 changes: 1 addition & 2 deletions common/src/test/java/org/astraea/common/ByteUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,9 +94,8 @@ void testReadAndToBytesClusterInfo() {
var bytes = ByteUtils.toBytes(clusterInfo);
Assertions.assertDoesNotThrow(() -> ByteUtils.readClusterInfo(bytes));
var deserializedClusterInfo = ByteUtils.readClusterInfo(bytes);

Assertions.assertEquals(clusterInfo.clusterId(), deserializedClusterInfo.clusterId());
Assertions.assertTrue(clusterInfo.nodes().containsAll(deserializedClusterInfo.nodes()));
Assertions.assertEquals(clusterInfo.nodes(), deserializedClusterInfo.nodes());
Assertions.assertEquals(clusterInfo.topics(), deserializedClusterInfo.topics());
Assertions.assertEquals(clusterInfo.replicas(), deserializedClusterInfo.replicas());
}
Expand Down