Tune kafka and grpc maximum message size (#2661)

alibaba · Apr 27, 2023 · 57dba68 · 57dba68
1 parent 1f0b8b9
commit 57dba68
Show file tree

Hide file tree

Showing 7 changed files with 14 additions and 14 deletions.
diff --git a/charts/graphscope-store/README.md b/charts/graphscope-store/README.md
@@ -88,6 +88,8 @@ Here we give a list of most frequently used parameters.
 | javaOpts | Java options | "" |
 | auth.username | Username | "" |
 | auth.password | Password | "" |
+| rpcMaxBytesMb | GRPC maximum message size | 4 |
+| kafkaProducerCustomConfigs | Kafka producer max request size | "max.request.size:1048576000" |
 
 
 ### Statefulset parameters

diff --git a/charts/graphscope-store/templates/configmap.yaml b/charts/graphscope-store/templates/configmap.yaml
@@ -48,6 +48,7 @@ data:
     ## Kafka Config
     kafka.servers=KAFKA_SERVERS
     kafka.topic={{ .Values.kafkaTopic }}
+    kafka.producer.custom.configs={{ .Values.kafkaProducerCustomConfigs }}
 
     ## Frontend Config
     gremlin.server.port=12312

diff --git a/charts/graphscope-store/values.yaml b/charts/graphscope-store/values.yaml
@@ -529,6 +529,7 @@ storeWriteThreadCount: 1
 
 ## Kafka Config
 kafkaTopic: "graphscope"
+kafkaProducerCustomConfigs: ""
 
 ## Frontend Config
 # gremlinServerPort: 12312

diff --git a/...e_engine/common/src/main/java/com/alibaba/graphscope/groot/common/config/KafkaConfig.java b/...e_engine/common/src/main/java/com/alibaba/graphscope/groot/common/config/KafkaConfig.java
@@ -26,6 +26,6 @@ public class KafkaConfig {
     public static final Config<String> KAFKA_PRODUCER_CUSTOM_CONFIGS =
             Config.stringConfig("kafka.producer.custom.configs", "");
 
-    public static final Config<Integer> KAFKA_MAX_MESSEAGE_MB =
+    public static final Config<Integer> KAFKA_MAX_MESSAGE_MB =
             Config.intConfig("kafka.max.message.mb", 20);
 }
diff --git a/...ve_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/rpc/ChannelManager.java b/...ve_engine/groot-module/src/main/java/com/alibaba/graphscope/groot/rpc/ChannelManager.java
@@ -54,7 +54,7 @@ public void start() {
             Map<Integer, ManagedChannel> idxToChannel =
                     this.roleToChannels.computeIfAbsent(role, k -> new HashMap<>());
             int count =
-                    Integer.valueOf(
+                    Integer.parseInt(
                             this.configs.get(
                                     String.format(CommonConfig.NODE_COUNT_FORMAT, role.getName()),
                                     "0"));
@@ -105,15 +105,11 @@ public void start() {
                 for (int i = 0; i < count; i++) {
                     String host = hostTemplate.replace("{}", String.valueOf(i));
                     logger.info(
-                            "create channel to role ["
-                                    + role.getName()
-                                    + "] #["
-                                    + i
-                                    + "]. host ["
-                                    + host
-                                    + "], port ["
-                                    + port
-                                    + "]");
+                            "Creating channel to role {} #{}, host {}, port {}",
+                            role.getName(),
+                            i,
+                            host,
+                            port);
                     ManagedChannel channel =
                             ManagedChannelBuilder.forAddress(host, port)
                                     .maxInboundMessageSize(this.rpcMaxBytes)
@@ -123,7 +119,7 @@ public void start() {
                 }
             } else {
                 for (int i = 0; i < count; i++) {
-                    logger.debug("create channel to role [" + role.getName() + "] #[" + i + "]");
+                    logger.info("Create channel to role {} #{}", role.getName(), i);
                     String uri = SCHEME + "://" + role.getName() + "/" + i;
                     ManagedChannel channel =
                             ManagedChannelBuilder.forTarget(uri)

diff --git a/...ne/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogService.java b/...ne/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogService.java
@@ -54,7 +54,7 @@ public KafkaLogService(Configs configs) {
         this.topic = KafkaConfig.KAKFA_TOPIC.get(configs);
         this.queueCount = CommonConfig.INGESTOR_QUEUE_COUNT.get(configs);
         this.replicationFactor = KafkaConfig.KAFKA_REPLICATION_FACTOR.get(configs);
-        this.maxMessageMb = KafkaConfig.KAFKA_MAX_MESSEAGE_MB.get(configs);
+        this.maxMessageMb = KafkaConfig.KAFKA_MAX_MESSAGE_MB.get(configs);
     }
 
     @Override

diff --git a/...ine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogWriter.java b/...ine/groot-module/src/main/java/com/alibaba/graphscope/groot/wal/kafka/KafkaLogWriter.java
@@ -57,7 +57,7 @@ public KafkaLogWriter(
     }
 
     @Override
-    public long append(LogEntry logEntry) throws IOException {
+    public long append(LogEntry logEntry) {
         Future<RecordMetadata> future =
                 producer.send(
                         new ProducerRecord<>(this.topicName, this.partitionId, null, logEntry));