Make max encoding message size configurable

ballista/executor/executor_config_spec.toml

@@ -132,6 +132,12 @@ type = "u32"
 default = "16777216"
 doc = "The maximum size of a decoded message at the grpc server side. Default: 16MB"
 
+[[param]]
+name = "grpc_server_max_encoding_message_size"
+type = "u32"
+default = "16777216"
+doc = "The maximum size of an encoded message at the grpc server side. Default: 16MB"
+
 [[param]]
 name = "executor_heartbeat_interval_seconds"
 type = "u64"

ballista/executor/src/bin/main.rs

@@ -79,6 +79,7 @@ async fn main() -> Result<()> {
         job_data_ttl_seconds: opt.job_data_ttl_seconds,
         job_data_clean_up_interval_seconds: opt.job_data_clean_up_interval_seconds,
         grpc_server_max_decoding_message_size: opt.grpc_server_max_decoding_message_size,
+        grpc_server_max_encoding_message_size: opt.grpc_server_max_encoding_message_size,
         executor_heartbeat_interval_seconds: opt.executor_heartbeat_interval_seconds,
         data_cache_policy: opt.data_cache_policy,
         cache_dir: opt.cache_dir,

ballista/executor/src/executor_process.rs

@@ -96,6 +96,8 @@ pub struct ExecutorProcessConfig {
     pub cache_io_concurrency: u32,
     /// The maximum size of a decoded message at the grpc server side.
     pub grpc_server_max_decoding_message_size: u32,
+    /// The maximum size of an encoded message at the grpc server side.
+    pub grpc_server_max_encoding_message_size: u32,
     pub executor_heartbeat_interval_seconds: u64,
     /// Optional execution engine to use to execute physical plans, will default to
     /// DataFusion if none is provided.

ballista/executor/src/executor_server.rs

@@ -112,6 +112,9 @@ pub async fn startup<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan>(
             BALLISTA_VERSION, addr
         );
         let server = ExecutorGrpcServer::new(executor_server.clone())
+            .max_encoding_message_size(
+                config.grpc_server_max_encoding_message_size as usize,
+            )
             .max_decoding_message_size(
                 config.grpc_server_max_decoding_message_size as usize,
             );

ballista/scheduler/scheduler_config_spec.toml

@@ -171,6 +171,12 @@ type = "u32"
 default = "16777216"
 doc = "The maximum size of a decoded message at the grpc server side. Default: 16MB"
 
+[[param]]
+name = "grpc_server_max_encoding_message_size"
+type = "u32"
+default = "16777216"
+doc = "The maximum size of an encoded message at the grpc server side. Default: 16MB"
+
 [[param]]
 name = "executor_timeout_seconds"
 type = "u64"

ballista/scheduler/src/bin/main.rs

@@ -155,6 +155,7 @@ async fn main() -> Result<()> {
         scheduler_event_expected_processing_duration: opt
             .scheduler_event_expected_processing_duration,
         grpc_server_max_decoding_message_size: opt.grpc_server_max_decoding_message_size,
+        grpc_server_max_encoding_message_size: opt.grpc_server_max_encoding_message_size,
         executor_timeout_seconds: opt.executor_timeout_seconds,
         expire_dead_executor_interval_seconds: opt.expire_dead_executor_interval_seconds,
     };

ballista/scheduler/src/config.rs

@@ -56,6 +56,8 @@ pub struct SchedulerConfig {
     pub scheduler_event_expected_processing_duration: u64,
     /// The maximum size of a decoded message at the grpc server side.
     pub grpc_server_max_decoding_message_size: u32,
+    /// The maximum size of an encoded message at the grpc server side.
+    pub grpc_server_max_encoding_message_size: u32,
     /// The executor timeout in seconds. It should be longer than executor's heartbeat intervals.
     pub executor_timeout_seconds: u64,
     /// The interval to check expired or dead executors
@@ -79,6 +81,7 @@ impl Default for SchedulerConfig {
             executor_termination_grace_period: 0,
             scheduler_event_expected_processing_duration: 0,
             grpc_server_max_decoding_message_size: 16777216,
+            grpc_server_max_encoding_message_size: 16777216,
             executor_timeout_seconds: 180,
             expire_dead_executor_interval_seconds: 15,
         }
@@ -167,6 +170,11 @@ impl SchedulerConfig {
         self.grpc_server_max_decoding_message_size = value;
         self
     }
+
+    pub fn with_grpc_server_max_encoding_message_size(mut self, value: u32) -> Self {
+        self.grpc_server_max_encoding_message_size = value;
+        self
+    }
 }
 
 #[derive(Clone, Debug)]

ballista/scheduler/src/scheduler_process.rs

@@ -75,6 +75,9 @@ pub async fn start_server(
             let config = &scheduler_server.state.config;
             let scheduler_grpc_server =
                 SchedulerGrpcServer::new(scheduler_server.clone())
+                    .max_encoding_message_size(
+                        config.grpc_server_max_encoding_message_size as usize,
+                    )
                     .max_decoding_message_size(
                         config.grpc_server_max_decoding_message_size as usize,
                     );