...

create temporary table table1 (
    val1 STRING
) WITH (
      'connector' = 'datagen',
      'number-of-rows' = '1'
);

create temporary table table2 (
    val2 STRING
) WITH (
      'connector' = 'datagen',
      'number-of-rows' = '1'
);

create temporary table table3 (
    val3 STRING
) WITH (
      'connector' = 'datagen',
      'number-of-rows' = '1'
);

DQL

select val1 from table1;

select val1, count(*) from table1 
	 group by val1;

select val1, count(*) from table1 
    left join table2 on val1=val2 
    left join table3 on val2=val3 
group by val1;

Benchmark

Notice: To test the performance bottleneck of Flink Client in interactive scenarios, we used a version of Flink Cluster running with more scheduling optimizations than the community(e.g. HA improvement mentioned in FLIP-403) as the baseline.

...

In high concurrency scenarios, the CPU usage of the Flink cluster was reduced by 20% with no performance degradation in QPS by using proper retry strategy configuration.

	Baseline(128 Concurrency)		Experiment(128 Concurrency)
Retry Strategy	DEFAULT_RETRY_MILLIS=100 ms		Optimized retry strategy
Join QPS	47		48
JM CPU Usage AVG	17.3

Image Removed

Image Added

14.5

Image Removed

Image Added
	TM CPU Usage AVG	9.0	7.2

Changed Public Interfaces

DeploymentOptions

CollectOptions

Proposed Change

Overview

• Reuse ClusterDescriptor when interacting with a dedicated session cluster
• Reuse ClientHighAvailabilityServices for leader retrieval with the same session cluster
• Reuse RestClusterClient when sending requests to the same session cluster
• Reuse the HTTP connections between JDBC Driver to SQL Gateway and Flink Client to Flink Cluster with the same target address
• Use ThreadLocal<ObjectMapper> instead of a global ObjectMapper instance
• Use a configurable retry strategy in CollectResultFetcher to reduce unnecessary workload

...

ClusterDescriptor is created by ClusterClientFactory. When users have enabled the 'execution.interactive-client' option, ClusterClientFactory will return a cached ClusterDescriptor for each ClusterID. The key and value in the cache would be ClusterID and ClusterDescriptor. The cache will be configured with an expireAfterAccess time.

ClientHighAvailabilityServices Reuse

...