diff --git a/mtr/binlog_streaming/include/diff_with_storage_object.inc b/mtr/binlog_streaming/include/diff_with_storage_object.inc
index bee576d..8925e0f 100644
--- a/mtr/binlog_streaming/include/diff_with_storage_object.inc
+++ b/mtr/binlog_streaming/include/diff_with_storage_object.inc
@@ -33,7 +33,7 @@ if ($storage_backend == file)
 if ($storage_backend == s3)
 {
   --let $downloaded_file_path = $MYSQL_TMP_DIR/diff_with_storage_object.downloaded
-  --exec $aws_cli s3 cp s3://$aws_s3_bucket$storage_object $downloaded_file_path > /dev/null
+  --exec $aws_cli s3 cp s3://$aws_s3_bucket/$storage_object $downloaded_file_path > /dev/null
   --diff_files $local_file $downloaded_file_path
   --remove_file $downloaded_file_path
 }
diff --git a/mtr/binlog_streaming/include/patch_binlog_in_use_flag.inc b/mtr/binlog_streaming/include/patch_binlog_in_use_flag.inc
new file mode 100644
index 0000000..f197c7e
--- /dev/null
+++ b/mtr/binlog_streaming/include/patch_binlog_in_use_flag.inc
@@ -0,0 +1,31 @@
+# Copy a binlog file and clear LOG_EVENT_BINLOG_IN_USE_F in the patched copy.
+#
+# Usage:
+#   --let $binlog_file_to_patch = $binlog_base_dir/$binlog_name
+#   --let PATCHED_BINLOG_FILE = $MYSQL_TMP_DIR/$binlog_name.patched
+#   --source ../include/patch_binlog_in_use_flag.inc
+
+--echo
+--echo *** Patching the server version of the second binlog file to clear the
+--echo *** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
+--copy_file $binlog_file_to_patch $PATCHED_BINLOG_FILE
+
+--perl
+  use strict;
+  use warnings;
+  use constant MAGIC_OFFSET => 21;
+  my $binlog_file_perl = $ENV{'PATCHED_BINLOG_FILE'};
+
+  open(my $fh, '+<:raw', $binlog_file_perl) or die "Failed to open file: $!";
+
+  seek($fh, MAGIC_OFFSET, 0);
+  my $byte;
+  read($fh, $byte, 1);
+
+  $byte = ord($byte) & 0xFE;
+
+  seek($fh, MAGIC_OFFSET, 0);
+  print $fh pack('C', $byte);
+
+  close($fh);
+EOF
diff --git a/mtr/binlog_streaming/include/set_up_binsrv_environment.inc b/mtr/binlog_streaming/include/set_up_binsrv_environment.inc
index 6f8d5e1..2785b9a 100644
--- a/mtr/binlog_streaming/include/set_up_binsrv_environment.inc
+++ b/mtr/binlog_streaming/include/set_up_binsrv_environment.inc
@@ -38,11 +38,11 @@ if ($storage_backend == file)
 if ($storage_backend == s3)
 {
   --let $binsrv_buffer_path = $MYSQL_TMP_DIR/buffer
-  --let $binsrv_storage_path = `SELECT CONCAT('/mtr-', UUID())`
+  --let $binsrv_storage_path = `SELECT CONCAT('mtr-', UUID())`
   --let $aws_s3_bucket = $MTR_BINSRV_AWS_S3_BUCKET
   if ($MTR_BINSRV_AWS_S3_ENDPOINT != '')
   {
-    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/$MTR_BINSRV_AWS_S3_BUCKET', '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('http://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$MTR_BINSRV_AWS_S3_ENDPOINT', '/$MTR_BINSRV_AWS_S3_BUCKET', '/$binsrv_storage_path');
   }
   if ($MTR_BINSRV_AWS_S3_ENDPOINT == '')
   {
@@ -51,7 +51,7 @@ if ($storage_backend == s3)
     {
       --let $qualified_bucket = $qualified_bucket.$MTR_BINSRV_AWS_S3_REGION
     }
-    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$qualified_bucket',  '$binsrv_storage_path');
+    eval SET @storage_uri = CONCAT('s3://', '$MTR_BINSRV_AWS_ACCESS_KEY_ID', ':', '$MTR_BINSRV_AWS_SECRET_ACCESS_KEY', '@', '$qualified_bucket',  '/$binsrv_storage_path');
   }
 }
 
diff --git a/mtr/binlog_streaming/include/tear_down_binsrv_environment.inc b/mtr/binlog_streaming/include/tear_down_binsrv_environment.inc
index 0f92706..2df1d63 100644
--- a/mtr/binlog_streaming/include/tear_down_binsrv_environment.inc
+++ b/mtr/binlog_streaming/include/tear_down_binsrv_environment.inc
@@ -6,7 +6,7 @@ if ($storage_backend == file)
 }
 if ($storage_backend == s3)
 {
-  --exec $aws_cli s3 rm s3://$aws_s3_bucket$binsrv_storage_path/ --recursive > /dev/null
+  --exec $aws_cli s3 rm s3://$aws_s3_bucket/$binsrv_storage_path/ --recursive > /dev/null
   --force-rmdir $binsrv_buffer_path
 }
 
diff --git a/mtr/binlog_streaming/include/terminate_binsrv.inc b/mtr/binlog_streaming/include/terminate_binsrv.inc
new file mode 100644
index 0000000..95f1bca
--- /dev/null
+++ b/mtr/binlog_streaming/include/terminate_binsrv.inc
@@ -0,0 +1,19 @@
+# Terminate the Binlog Server Utility and wait until the process exits.
+#
+# Usage:
+#   --let $binsrv_pid = <pid>
+#   --source ../include/terminate_binsrv.inc
+
+--exec kill -s TERM $binsrv_pid
+--let EXPORTED_BINSRV_PID = $binsrv_pid
+--perl
+  use strict;
+  use warnings;
+  use Errno;
+  my $pid = $ENV{'EXPORTED_BINSRV_PID'};
+  my $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
+  while (!$not_present) {
+    sleep(1);
+    $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
+  }
+EOF
diff --git a/mtr/binlog_streaming/include/trigger_binsrv_pull.inc b/mtr/binlog_streaming/include/trigger_binsrv_pull.inc
new file mode 100644
index 0000000..d34499c
--- /dev/null
+++ b/mtr/binlog_streaming/include/trigger_binsrv_pull.inc
@@ -0,0 +1,13 @@
+# Start the Binlog Server Utility pull command in background.
+#
+# Usage:
+#   --let $binsrv_spawn_cmd_line = $BINSRV pull ... & echo \$! > $binsrv_pid_file
+#   --source ../include/trigger_binsrv_pull.inc
+
+--let EXPORTED_BINSRV_SPAWN_CMD_LINE = $binsrv_spawn_cmd_line
+--perl
+  use strict;
+  use warnings;
+  my $cmd = $ENV{'EXPORTED_BINSRV_SPAWN_CMD_LINE'};
+  system("$cmd");
+EOF
diff --git a/mtr/binlog_streaming/r/binsrv.result b/mtr/binlog_streaming/r/binsrv.result
index 398be76..3c845c1 100644
--- a/mtr/binlog_streaming/r/binsrv.result
+++ b/mtr/binlog_streaming/r/binsrv.result
@@ -29,6 +29,10 @@ INSERT INTO t1 VALUES(DEFAULT);
 *** Checking that the Binlog Server utility detected an empty storage
 include/assert_grep.inc [Binlog storage must be initialized on an empty directory]
 
+*** Verifying that binlog files were written to storage
+include/assert_grep.inc [First binlog should exist in storage]
+include/assert_grep.inc [Second binlog also hould exist in storage]
+
 *** Comparing server and downloaded versions of the first binlog file.
 
 *** Patching the server version of the second binlog file to clear the
diff --git a/mtr/binlog_streaming/r/checkpointing_verification.result b/mtr/binlog_streaming/r/checkpointing_verification.result
new file mode 100644
index 0000000..a14c220
--- /dev/null
+++ b/mtr/binlog_streaming/r/checkpointing_verification.result
@@ -0,0 +1,116 @@
+*** Testing checkpointing verification
+*** For S3 backend, each checkpoint triggers a full object re-upload.
+
+*** Resetting replication at the very beginning of the test.
+
+*** Creating a table and generating large binlog events
+CREATE TABLE t1(
+id SERIAL,
+val CHAR(64) CHARACTER SET ascii NOT NULL,
+PRIMARY KEY(id)
+) ENGINE=InnoDB;
+
+*** Generating data to trigger checkpointing
+
+*** Determining current open binlog file name
+
+*** Setting up checkpointing configuration
+
+*** Setting up configuration
+SET @save_binsrv_checkpoint_size = 'binsrv_checkpoint_size';
+SET @save_binsrv_checkpoint_interval = 'binsrv_checkpoint_interval';
+
+*** Test 1: Test checkpointing with existing open binary log
+*** Expectation:
+*** - Without checkpointing: storage object stays at magic payload size while binsrv runs
+*** - With size checkpointing: storage object grows beyond magic payload while binsrv runs
+*** Note: Checkpointing only works with S3 backend, not with file backend
+
+Executing binlog_server in pull mode without checkpointing
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+include/read_file_to_var.inc
+include/read_file_to_var.inc
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+
+Executing binlog_server in pull mode with checkpointing
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+include/read_file_to_var.inc
+include/read_file_to_var.inc
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+*** Test 2: Test that new updates in binary log on server will be flushed
+*** to the binlog_server storage directory due to checkpointing.
+
+
+*** Starting pull mode in background to monitor checkpointing
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+include/read_file_to_var.inc
+
+*** Capturing current storage object size (before workload)
+include/read_file_to_var.inc
+
+*** Generating more data to trigger checkpointing during pull
+
+*** Waiting a bit to allow checkpointing to occur
+
+*** Capturing storage object size after checkpointing workload
+include/read_file_to_var.inc
+
+*** Verifying checkpointing occurred (by storage object size change)
+
+*** Patching the server version of the binlog file to clear the
+*** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
+*** (The binlog is still open on the server, so we patch it instead of flushing
+*** to avoid adding a ROTATE event that wouldn't exist in the storage object)
+
+*** Terminating pull mode
+
+*** Verifying binlog file integrity after checkpointing
+
+*** Starting pull mode again to verify it resumes correctly without any issues.
+include/read_file_to_var.inc
+
+*** Adding more data
+INSERT INTO t1(val) VALUES(SHA2(LAST_INSERT_ID(), 256));
+FLUSH BINARY LOGS;
+
+*** Waiting a bit then terminating
+
+*** Verifying that resume worked correctly (no duplicate data)
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t1;
diff --git a/mtr/binlog_streaming/r/pull_mode.result b/mtr/binlog_streaming/r/pull_mode.result
index 7d9e95e..89153e1 100644
--- a/mtr/binlog_streaming/r/pull_mode.result
+++ b/mtr/binlog_streaming/r/pull_mode.result
@@ -60,4 +60,61 @@ FLUSH BINARY LOGS;
 *** Removing the Binlog Server utility log file.
 
 *** Removing the Binlog Server utility configuration file.
-KILL CONNECTION <CONNECTION_ID>;
+
+*** Pull mode: zero idle_time
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Starting Binlog Server Utility in background in pull mode with zero idle time
+include/read_file_to_var.inc
+
+*** Creating data.
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+*** Waiting for read timeout to trigger reconnection.
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Terminating the utility
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+
+*** Pull mode: very short timeouts and rapid reconnection
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Starting Binlog Server Utility in background
+include/read_file_to_var.inc
+
+*** Creating data and restarting server multiple times
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+# restart
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+# restart
+INSERT INTO t1 VALUES(DEFAULT);
+
+*** Terminating the utility
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t1;
diff --git a/mtr/binlog_streaming/r/read_timeout_flush.result b/mtr/binlog_streaming/r/read_timeout_flush.result
new file mode 100644
index 0000000..bb83ccc
--- /dev/null
+++ b/mtr/binlog_streaming/r/read_timeout_flush.result
@@ -0,0 +1,84 @@
+
+*** Test : Storage flush after read_timeout disconnection (PS-10320 fix)
+*** This test verifies that local binary log data is flushed to storage
+*** upon disconnection due to read_timeout
+*** This test also verifies existence of temporary files in buffer directory
+*** in S3 storage mode while binlog_server run is in progress.
+SET SESSION binlog_transaction_compression=ON;
+CREATE TABLE t2(id INT UNSIGNED NOT NULL AUTO_INCREMENT, val CHAR(64), PRIMARY KEY(id)) ENGINE=InnoDB;
+
+*** Starting with a fresh binlog
+FLUSH BINARY LOGS;
+
+*** Setting up binsrv with short read_timeout to trigger timeout
+
+*** Generating a configuration file in JSON format for the Binlog
+*** Server utility.
+
+*** Determining binlog file directory from the server.
+
+*** Creating a temporary directory <BINSRV_STORAGE_PATH> for storing
+*** binlog files downloaded via the Binlog Server utility.
+
+*** Starting Binlog Server Utility in background in pull mode
+include/read_file_to_var.inc
+
+*** Waiting for binsrv to initialize and create binlog file in storage
+
+*** Step 1: Check initial binlog size (baseline)
+include/read_file_to_var.inc
+
+*** Step 2: Execute INSERT and wait for read_timeout to trigger flush (read_timeout=20, wait 25 seconds)
+INSERT INTO t2(val) VALUES('test_data_before_timeout');
+
+*** Step 3: Check binlog size after read_timeout flush (should have increased)
+include/read_file_to_var.inc
+
+*** Verifying that data was flushed to storage after first read_timeout (PS-10320 fix)
+
+*** Step 4: Wait around idle_time and generate new data in the binary log. verify either in idle_time or upon reconnection data was not flushed to storage.
+INSERT INTO t2(val) VALUES('test_data_after_timeout');
+SET @save_binlog_row_value_options = @@GLOBAL.binlog_row_value_options;
+SET GLOBAL binlog_row_value_options = PARTIAL_JSON;
+Warnings:
+Warning	3647	When binlog_row_image=FULL, the option binlog_row_value_options=PARTIAL_JSON will be used only for the after-image. Full values will be written in the before-image, so the saving in disk space due to binlog_row_value_options is limited to less than 50%.
+UPDATE t2 set val='update_test_data_after_timeout' where val='test_data_after_timeout';
+CREATE TABLE tj (j JSON);
+INSERT INTO tj VALUES('{"name":"john","age":30,"city":"blr"}');
+INSERT INTO tj VALUES('{"arr":[1,2,3],"details":{"salary":5000}}');
+UPDATE tj SET j = JSON_REPLACE(j, '$.city', 'mysore') WHERE JSON_EXTRACT(j, '$.name') = 'john';
+UPDATE tj SET j = JSON_REPLACE(j, '$.arr[1]', 999);
+DROP TABLE tj;
+SET GLOBAL binlog_row_value_options = @save_binlog_row_value_options;
+
+*** Step 5: Check binlog size after around idle time (should NOT have changed - since no flush)
+include/read_file_to_var.inc
+
+*** Verifying that size did NOT change after reconnection (no flush)
+
+*** Step 6: Execute INSERT again
+INSERT INTO t2(val) VALUES('test_data_after_timeout');
+
+*** Step 7: Check size immediately after INSERT ie around idle/reconnection time (should NOT have changed - data in buffer)
+include/read_file_to_var.inc
+
+*** Verifying that size did NOT change immediately after INSERT (data still in buffer)
+
+*** Step 8: Wait for read_timeout to trigger flush again (read_timeout=20)
+
+*** Step 9: Check binlog size after second read_timeout flush (should have increased)
+include/read_file_to_var.inc
+
+*** Verifying that size increased after second read_timeout flush
+FLUSH BINARY LOGS;
+
+*** Terminating the utility
+
+*** Final verification: Compare storage object with server binlog
+
+*** Removing the Binlog Server utility storage directory.
+
+*** Removing the Binlog Server utility log file.
+
+*** Removing the Binlog Server utility configuration file.
+DROP TABLE t2;
diff --git a/mtr/binlog_streaming/t/binsrv.test b/mtr/binlog_streaming/t/binsrv.test
index 5fa6e0c..68fa72c 100644
--- a/mtr/binlog_streaming/t/binsrv.test
+++ b/mtr/binlog_streaming/t/binsrv.test
@@ -60,6 +60,38 @@ INSERT INTO t1 VALUES(DEFAULT);
 --let $assert_select = binlog storage initialized on an empty directory
 --source include/assert_grep.inc
 
+--echo
+--echo *** Verifying that binlog files were written to storage
+--let $assert_text = First binlog should exist in storage
+--let $assert_file = $MYSQL_TMP_DIR/storage_check.log
+if ($storage_backend == file)
+{
+  --exec ls -la $binsrv_storage_path/$first_binlog* > $assert_file 2>&1 || true
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3 ls s3://$aws_s3_bucket/$binsrv_storage_path/$first_binlog > $assert_file 2>&1 || true
+}
+--let $assert_count = 2
+--let $assert_select = $first_binlog
+--source include/assert_grep.inc
+--remove_file $assert_file
+
+--let $assert_text = Second binlog also hould exist in storage
+--let $assert_file = $MYSQL_TMP_DIR/storage_check.log
+if ($storage_backend == file)
+{
+  --exec ls -la $binsrv_storage_path/$second_binlog* > $assert_file 2>&1 || true
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3 ls s3://$aws_s3_bucket/$binsrv_storage_path/$second_binlog > $assert_file 2>&1 || true
+}
+--let $assert_count = 2
+--let $assert_select = $second_binlog
+--source include/assert_grep.inc
+--remove_file $assert_file
+
 # At this point we have 2 binlog files $first_binlog (already closed/rotated
 # by the server) and $second_binlog (currently open).
 
@@ -77,32 +109,9 @@ INSERT INTO t1 VALUES(DEFAULT);
 # "\xFEbin" + 17, the offset of the 'flags' field in the common header).
 # So, here we create a copy of the second binlog file, patch this byte and
 # perform diff on this patched copy.
-
---echo
---echo *** Patching the server version of the second binlog file to clear the
---echo *** LOG_EVENT_BINLOG_IN_USE_F (currently in use) flag.
+--let $binlog_file_to_patch = $binlog_base_dir/$second_binlog
 --let PATCHED_BINLOG_FILE = $MYSQL_TMP_DIR/$second_binlog.patched
---copy_file $binlog_base_dir/$second_binlog $PATCHED_BINLOG_FILE
-
---perl
-  use strict;
-  use warnings;
-  use constant MAGIC_OFFSET => 21;
-  my $binlog_file_perl = $ENV{'PATCHED_BINLOG_FILE'};
-
-  open(my $fh, '+<:raw', $binlog_file_perl) or die "Failed to open file: $!";
-
-  seek($fh, MAGIC_OFFSET, 0);
-  my $byte;
-  read($fh, $byte, 1);
-
-  $byte = ord($byte) & 0xFE;
-
-  seek($fh, MAGIC_OFFSET, 0);
-  print $fh pack('C', $byte);
-
-  close($fh);
-EOF
+--source ../include/patch_binlog_in_use_flag.inc
 
 --echo
 --echo *** Comparing server and downloaded versions of the second binlog file.
diff --git a/mtr/binlog_streaming/t/checkpointing_verification-master.opt b/mtr/binlog_streaming/t/checkpointing_verification-master.opt
new file mode 100644
index 0000000..c7529f0
--- /dev/null
+++ b/mtr/binlog_streaming/t/checkpointing_verification-master.opt
@@ -0,0 +1,2 @@
+--gtid-mode=on
+--enforce-gtid-consistency
diff --git a/mtr/binlog_streaming/t/checkpointing_verification.combinations b/mtr/binlog_streaming/t/checkpointing_verification.combinations
new file mode 100644
index 0000000..a014f18
--- /dev/null
+++ b/mtr/binlog_streaming/t/checkpointing_verification.combinations
@@ -0,0 +1,8 @@
+[checkpointing_size]
+init-connect = SET @binsrv_checkpointing = 'size'
+
+[checkpointing_interval]
+init-connect = SET @binsrv_checkpointing = 'interval'
+
+[checkpointing_both]
+init-connect = SET @binsrv_checkpointing = 'both'
diff --git a/mtr/binlog_streaming/t/checkpointing_verification.test b/mtr/binlog_streaming/t/checkpointing_verification.test
new file mode 100644
index 0000000..68353aa
--- /dev/null
+++ b/mtr/binlog_streaming/t/checkpointing_verification.test
@@ -0,0 +1,329 @@
+--source ../include/have_binsrv.inc
+--source include/big_test.inc
+
+--source ../include/v80_v84_compatibility_defines.inc
+
+--source include/count_sessions.inc
+
+# identifying backend storage type ('file' or 's3')
+--source ../include/identify_storage_backend.inc
+# identifying interval checkpointing mode from the combination
+--let $extracted_init_connect_variable_name = binsrv_checkpointing
+--source ../include/extract_init_connect_variable_value.inc
+
+--echo *** Testing checkpointing verification
+--echo *** For S3 backend, each checkpoint triggers a full object re-upload.
+
+--echo
+--echo *** Resetting replication at the very beginning of the test.
+--disable_query_log
+eval $stmt_reset_binary_logs_and_gtids;
+--enable_query_log
+
+--echo
+--echo *** Creating a table and generating large binlog events
+CREATE TABLE t1(
+  id SERIAL,
+  val CHAR(64) CHARACTER SET ascii NOT NULL,
+  PRIMARY KEY(id)
+) ENGINE=InnoDB;
+
+--echo
+--echo *** Generating data to trigger checkpointing
+--disable_query_log
+--let $transaction_idx = 0
+while ($transaction_idx < 20)
+{
+  START TRANSACTION;
+  --let $statement_idx = 0
+  while ($statement_idx < 100)
+  {
+    eval INSERT INTO t1(val) VALUES(SHA2(LAST_INSERT_ID(), 256));
+    --inc $statement_idx
+  }
+  COMMIT;
+  --inc $transaction_idx
+}
+--enable_query_log
+
+--echo
+--echo *** Determining current open binlog file name
+--let $binlog_name = query_get_value($stmt_show_binary_log_status, File, 1)
+
+--let $binsrv_replication_mode = position
+
+--echo
+--echo *** Setting up checkpointing configuration
+
+if ($extracted_init_connect_variable_value == size)
+{
+  --let $binsrv_checkpoint_size = 1K
+  --let $binsrv_checkpoint_interval = 0
+}
+if ($extracted_init_connect_variable_value == interval)
+{
+  --let $binsrv_checkpoint_interval = 2s
+  --let $binsrv_checkpoint_size = 0
+}
+if ($extracted_init_connect_variable_value == both)
+{
+  --let $binsrv_checkpoint_size = 1K
+  --let $binsrv_checkpoint_interval = 2s
+}
+
+--echo
+--echo *** Setting up configuration
+--let $binsrv_connect_timeout = 10
+--let $binsrv_read_timeout = 200
+--let $binsrv_idle_time = 1
+--let $binsrv_verify_checksum = TRUE
+
+--replace_result $binsrv_checkpoint_size binsrv_checkpoint_size
+--eval SET @save_binsrv_checkpoint_size = '$binsrv_checkpoint_size'
+--replace_result $binsrv_checkpoint_interval binsrv_checkpoint_interval
+--eval SET @save_binsrv_checkpoint_interval = '$binsrv_checkpoint_interval'
+
+--echo
+--echo *** Test 1: Test checkpointing with existing open binary log
+--echo *** Expectation:
+--echo *** - Without checkpointing: storage object stays at magic payload size while binsrv runs
+--echo *** - With size checkpointing: storage object grows beyond magic payload while binsrv runs
+--echo *** Note: Checkpointing only works with S3 backend, not with file backend
+
+--echo
+--echo Executing binlog_server in pull mode without checkpointing
+--let $binsrv_checkpoint_size =
+--let $binsrv_checkpoint_interval =
+--let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_no_ckpt.pid
+--source ../include/set_up_binsrv_environment.inc
+
+--let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > $binsrv_log_path 2>&1 & echo \$! > $binsrv_pid_file
+--source ../include/trigger_binsrv_pull.inc
+
+--let $read_from_file = $binsrv_pid_file
+--source include/read_file_to_var.inc
+--let $binsrv_pid = $result
+
+--sleep 10
+
+--let $size_file = $MYSQL_TMP_DIR/binsrv_no_ckpt_size
+
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$binlog_name' > '$size_file'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$binlog_name --query ContentLength --output text > $size_file 2>/dev/null || echo 0 > $size_file
+}
+
+--let $read_from_file = $size_file
+--source include/read_file_to_var.inc
+--let $no_ckpt_size = $result
+if ($no_ckpt_size != 4)
+{
+  --echo Binlog size is $no_ckpt_size
+  --die Binlog size to be at magic payload before flush
+}
+
+--remove_file $size_file
+
+--source ../include/terminate_binsrv.inc
+--remove_file $binsrv_pid_file
+
+--source ../include/tear_down_binsrv_environment.inc
+
+
+--echo
+--echo Executing binlog_server in pull mode with checkpointing
+--let $binsrv_checkpoint_size = `select @save_binsrv_checkpoint_size`
+--let $binsrv_checkpoint_interval =`select @save_binsrv_checkpoint_interval`
+--source ../include/set_up_binsrv_environment.inc
+
+--let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_size_ckpt.pid
+--let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > $binsrv_log_path 2>&1 & echo \$! > $binsrv_pid_file
+--source ../include/trigger_binsrv_pull.inc
+
+--let $read_from_file = $binsrv_pid_file
+--source include/read_file_to_var.inc
+--let $binsrv_pid = $result
+
+--sleep 10
+
+--let $size_file = $MYSQL_TMP_DIR/binsrv_size_ckpt_size
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$binlog_name' > '$size_file'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$binlog_name --query ContentLength --output text > $size_file 2>/dev/null || echo 0 > $size_file
+}
+--let $read_from_file = $size_file
+--source include/read_file_to_var.inc
+--let $size_ckpt_size = $result
+--remove_file $size_file
+--let $size_ckpt_grew = `SELECT ($size_ckpt_size > $no_ckpt_size)`
+if ($binsrv_checkpoint_size == 1K)
+{
+  if ($size_ckpt_grew != 1)
+  {
+    --die expected storage object to grow with size checkpointing
+  }
+}
+if ($binsrv_checkpoint_size != 1K)
+{
+  if ($size_ckpt_size != 4)
+  {
+    --die expected storage object size to be at magic payload without size checkpointing
+  }
+}
+--source ../include/terminate_binsrv.inc
+--remove_file $binsrv_pid_file
+--source ../include/tear_down_binsrv_environment.inc
+
+--echo *** Test 2: Test that new updates in binary log on server will be flushed
+--echo *** to the binlog_server storage directory due to checkpointing.
+--echo
+
+--echo
+--echo *** Starting pull mode in background to monitor checkpointing
+--let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_checkpointing.pid
+--source ../include/set_up_binsrv_environment.inc
+--let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > $binsrv_log_path 2>&1 & echo \$! > $binsrv_pid_file
+--source ../include/trigger_binsrv_pull.inc
+
+--let $read_from_file = $binsrv_pid_file
+--source include/read_file_to_var.inc
+--let $binsrv_pid = $result
+
+--sleep 10
+
+--echo
+--echo *** Capturing current storage object size (before workload)
+
+--let $size_file = $MYSQL_TMP_DIR/binsrv_storage_object_size_before
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$binlog_name' > '$size_file'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$binlog_name --query ContentLength --output text > $size_file 2>/dev/null || echo 0 > $size_file
+}
+--let $read_from_file = $size_file
+--source include/read_file_to_var.inc
+--let $storage_object_size_before = $result
+--remove_file $size_file
+
+--echo
+--echo *** Generating more data to trigger checkpointing during pull
+--disable_query_log
+--let $transaction_idx = 0
+while ($transaction_idx < 20)
+{
+  START TRANSACTION;
+  --let $statement_idx = 0
+  while ($statement_idx < 50)
+  {
+    eval INSERT INTO t1(val) VALUES(SHA2(LAST_INSERT_ID(), 256));
+    --inc $statement_idx
+  }
+  COMMIT;
+  --inc $transaction_idx
+}
+--enable_query_log
+
+--echo
+--echo *** Waiting a bit to allow checkpointing to occur
+--sleep 10
+
+--echo
+--echo *** Capturing storage object size after checkpointing workload
+--let $size_file = $MYSQL_TMP_DIR/binsrv_storage_object_size_after
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$binlog_name' > '$size_file'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$binlog_name --query ContentLength --output text > $size_file 2>/dev/null || echo 0 > $size_file
+}
+--let $read_from_file = $size_file
+--source include/read_file_to_var.inc
+--let $storage_object_size_after = $result
+--remove_file $size_file
+
+--echo
+--echo *** Verifying checkpointing occurred (by storage object size change)
+if ($extracted_init_connect_variable_value == size)
+{
+  if ($storage_backend == s3)
+  {
+    --let $size_grew = `SELECT ($storage_object_size_after > $storage_object_size_before)`
+    if ($size_grew != 1)
+    {
+      --die storage object size did not grow after checkpointing workload
+    }
+  }
+}
+--let $binlog_file_to_patch = $binlog_base_dir/$binlog_name
+--let PATCHED_BINLOG_FILE = $MYSQL_TMP_DIR/$binlog_name.patched
+--source ../include/patch_binlog_in_use_flag.inc
+
+--echo
+--echo *** Terminating pull mode
+--source ../include/terminate_binsrv.inc
+
+--remove_file $binsrv_pid_file
+
+--echo
+--echo *** Verifying binlog file integrity after checkpointing
+--let $local_file = $PATCHED_BINLOG_FILE
+--let $storage_object = $binsrv_storage_path/$binlog_name
+--source ../include/diff_with_storage_object.inc
+
+--remove_file $PATCHED_BINLOG_FILE
+
+--echo
+--echo *** Starting pull mode again to verify it resumes correctly without any issues.
+--let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_checkpointing_resume.pid
+--let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > $binsrv_log_path 2>&1 & echo \$! > $binsrv_pid_file
+--source ../include/trigger_binsrv_pull.inc
+
+--let $read_from_file = $binsrv_pid_file
+--source include/read_file_to_var.inc
+--let $binsrv_pid = $result
+
+--echo
+--echo *** Adding more data
+eval INSERT INTO t1(val) VALUES(SHA2(LAST_INSERT_ID(), 256));
+FLUSH BINARY LOGS;
+
+--echo
+--echo *** Waiting a bit then terminating
+--sleep 10
+--source ../include/terminate_binsrv.inc
+--remove_file $binsrv_pid_file
+
+--echo
+--echo *** Verifying that resume worked correctly (no duplicate data)
+--let $local_file = $binlog_base_dir/$binlog_name
+--let $storage_object = $binsrv_storage_path/$binlog_name
+--source ../include/diff_with_storage_object.inc
+
+--source ../include/tear_down_binsrv_environment.inc
+
+# Clean up any remaining binlog dump connections
+--let $binlog_dump_connection_id = `SELECT ID FROM performance_schema.processlist WHERE COMMAND = 'Binlog Dump'`
+if ($binlog_dump_connection_id != '')
+{
+  --disable_query_log
+  --replace_result $binlog_dump_connection_id <CONNECTION_ID>
+  eval KILL CONNECTION $binlog_dump_connection_id;
+  --enable_query_log
+}
+
+--source include/wait_until_count_sessions.inc
+
+DROP TABLE t1;
diff --git a/mtr/binlog_streaming/t/pull_mode.test b/mtr/binlog_streaming/t/pull_mode.test
index e5c2ccb..9dd195e 100644
--- a/mtr/binlog_streaming/t/pull_mode.test
+++ b/mtr/binlog_streaming/t/pull_mode.test
@@ -27,13 +27,7 @@ eval $stmt_reset_binary_logs_and_gtids;
 --let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_utility.pid
 --let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > /dev/null & echo \$! > $binsrv_pid_file
 
---let EXPORTED_BINSRV_SPAWN_CMD_LINE = $binsrv_spawn_cmd_line
---perl
-  use strict;
-  use warnings;
-  my $cmd = $ENV{'EXPORTED_BINSRV_SPAWN_CMD_LINE'};
-  system("$cmd");
-EOF
+--source ../include/trigger_binsrv_pull.inc
 
 --let $read_from_file = $binsrv_pid_file
 --source include/read_file_to_var.inc
@@ -122,21 +116,7 @@ if ($grep_status != 0)
 --echo
 --echo *** Sending SIGTERM signal to the Binlog Server Utility and waiting for
 --echo *** the process to terminate
---replace_result $binsrv_pid <BINSRV_PID>
---exec kill -s TERM $binsrv_pid
---let EXPORTED_BINSRV_PID = $binsrv_pid
-
---perl
-  use strict;
-  use warnings;
-  use Errno;
-  my $pid = $ENV{'EXPORTED_BINSRV_PID'};
-  my $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
-  while (!$not_present) {
-    sleep(1);
-    $not_present = (!kill(0, $pid) && $! == Errno::ESRCH);
-  }
-EOF
+--source ../include/terminate_binsrv.inc
 
 --remove_file $binsrv_pid_file
 
@@ -161,12 +141,104 @@ EOF
 # cleaning up
 --source ../include/tear_down_binsrv_environment.inc
 
+--echo
+--echo *** Pull mode: zero idle_time
+--let $binsrv_connect_timeout = 3
+--let $binsrv_read_timeout = 3
+--let $binsrv_idle_time = 0
+--let $binsrv_verify_checksum = TRUE
+--let $binsrv_replication_mode = position
+--source ../include/set_up_binsrv_environment.inc
+
+--echo
+--echo *** Starting Binlog Server Utility in background in pull mode with zero idle time
+--let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_pull_zero_idle.pid
+--let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > /dev/null & echo \$! > $binsrv_pid_file
+
+--source ../include/trigger_binsrv_pull.inc
+
+--let $read_from_file = $binsrv_pid_file
+--source include/read_file_to_var.inc
+--let $binsrv_pid = $result
+
+--echo
+--echo *** Creating data.
+CREATE TABLE t1(id INT UNSIGNED NOT NULL AUTO_INCREMENT, PRIMARY KEY(id)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES(DEFAULT);
+
+--echo *** Waiting for read timeout to trigger reconnection.
+--sleep 4
+INSERT INTO t1 VALUES(DEFAULT);
+
+--echo
+--echo *** Terminating the utility
+--sleep 2
+--source ../include/terminate_binsrv.inc
+
+--remove_file $binsrv_pid_file
+
+--source ../include/tear_down_binsrv_environment.inc
+
+--echo
+--echo *** Pull mode: very short timeouts and rapid reconnection
+--let $binsrv_connect_timeout = 1
+--let $binsrv_read_timeout = 1
+--let $binsrv_idle_time = 1
+--let $binsrv_verify_checksum = TRUE
+--let $binsrv_replication_mode = position
+--source ../include/set_up_binsrv_environment.inc
+
+--echo
+--echo *** Starting Binlog Server Utility in background
+--let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_pull_short_timeout.pid
+--let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > /dev/null & echo \$! > $binsrv_pid_file
+
+--source ../include/trigger_binsrv_pull.inc
+
+--let $read_from_file = $binsrv_pid_file
+--source include/read_file_to_var.inc
+--let $binsrv_pid = $result
+
+--echo
+--echo *** Creating data and restarting server multiple times
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+
+--source include/shutdown_mysqld.inc
+--sleep 2
+--source include/start_mysqld.inc
+
+INSERT INTO t1 VALUES(DEFAULT);
+FLUSH BINARY LOGS;
+
+--source include/shutdown_mysqld.inc
+--sleep 2
+--source include/start_mysqld.inc
+
+INSERT INTO t1 VALUES(DEFAULT);
+
+--echo
+--echo *** Terminating the utility
+--source ../include/terminate_binsrv.inc
+
+--remove_file $binsrv_pid_file
+
+--source ../include/tear_down_binsrv_environment.inc
+
+DROP TABLE t1;
+
 # As the Binlog Server Utility interrupts the connection upon timeout, here we
 # need to close it on the MySQL server side as well in order to make sure that
 # MTR 'check-test' before and after the test produces the same output.
 --let $binlog_dump_connection_id = `SELECT ID FROM performance_schema.processlist WHERE COMMAND = 'Binlog Dump'`
---replace_result $binlog_dump_connection_id <CONNECTION_ID>
-eval KILL CONNECTION $binlog_dump_connection_id;
+if ($binlog_dump_connection_id != '')
+{
+  --disable_query_log
+  --disable_result_log
+  eval KILL CONNECTION $binlog_dump_connection_id;
+  --enable_result_log
+  --enable_query_log
+}
 
 # Also, we use 'count_sessions' include files to make sure that 'Binlog Dump'
 # connection is indeed closed.
diff --git a/mtr/binlog_streaming/t/read_timeout_flush-master.opt b/mtr/binlog_streaming/t/read_timeout_flush-master.opt
new file mode 100644
index 0000000..878a941
--- /dev/null
+++ b/mtr/binlog_streaming/t/read_timeout_flush-master.opt
@@ -0,0 +1,3 @@
+--gtid-mode=on
+--enforce-gtid-consistency
+--log-slave-updates
diff --git a/mtr/binlog_streaming/t/read_timeout_flush.test b/mtr/binlog_streaming/t/read_timeout_flush.test
new file mode 100644
index 0000000..22befbe
--- /dev/null
+++ b/mtr/binlog_streaming/t/read_timeout_flush.test
@@ -0,0 +1,264 @@
+--source ../include/have_binsrv.inc
+
+--source ../include/v80_v84_compatibility_defines.inc
+
+--source include/count_sessions.inc
+
+# identifying backend storage type ('file' or 's3')
+--source ../include/identify_storage_backend.inc
+
+--echo
+--echo *** Test : Storage flush after read_timeout disconnection (PS-10320 fix)
+--echo *** This test verifies that local binary log data is flushed to storage
+--echo *** upon disconnection due to read_timeout
+--echo *** This test also verifies existence of temporary files in buffer directory
+--echo *** in S3 storage mode while binlog_server run is in progress.
+--disable_query_log
+eval $stmt_reset_binary_logs_and_gtids;
+--enable_query_log
+
+# To test compressed events with binlog_server
+SET SESSION binlog_transaction_compression=ON;
+
+CREATE TABLE t2(id INT UNSIGNED NOT NULL AUTO_INCREMENT, val CHAR(64), PRIMARY KEY(id)) ENGINE=InnoDB;
+
+--echo
+--echo *** Starting with a fresh binlog
+FLUSH BINARY LOGS;
+--let $test_binlog = query_get_value($stmt_show_binary_log_status, File, 1)
+
+--echo
+--echo *** Setting up binsrv with short read_timeout to trigger timeout
+--let $binsrv_connect_timeout = 10
+--let $binsrv_read_timeout = 20
+--let $binsrv_idle_time = 30
+--let $binsrv_verify_checksum = TRUE
+--let $binsrv_replication_mode = position
+--source ../include/set_up_binsrv_environment.inc
+
+--echo
+--echo *** Starting Binlog Server Utility in background in pull mode
+--let $binsrv_pid_file = $MYSQL_TMP_DIR/binsrv_read_timeout_test.pid
+--let $binsrv_spawn_cmd_line = $BINSRV pull $binsrv_config_file_path > /dev/null & echo \$! > $binsrv_pid_file
+
+--source ../include/trigger_binsrv_pull.inc
+
+--let $read_from_file = $binsrv_pid_file
+--source include/read_file_to_var.inc
+--let $binsrv_pid = $result
+
+--echo
+--echo *** Waiting for binsrv to initialize and create binlog file in storage
+--sleep 4
+
+--echo
+--echo *** Step 1: Check initial binlog size (baseline)
+--let $size_file_initial = $MYSQL_TMP_DIR/storage_size_initial.txt
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$test_binlog' > '$size_file_initial'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$test_binlog --query ContentLength --output text > $size_file_initial 2>&1 || echo 0 > $size_file_initial
+}
+--let $read_from_file = $size_file_initial
+--source include/read_file_to_var.inc
+--let $size_initial = $result
+--remove_file $size_file_initial
+
+--echo
+--echo *** Step 2: Execute INSERT and wait for read_timeout to trigger flush (read_timeout=20, wait 25 seconds)
+INSERT INTO t2(val) VALUES('test_data_before_timeout');
+
+--disable_result_log
+if ($storage_backend == s3)
+{
+  --let $assert_text = Buffer directory should contain temporary files (UUID format)
+  --let $assert_file = $MYSQL_TMP_DIR/buffer_check.log
+  --disable_result_log
+  --exec ls -1 $binsrv_buffer_path 2>&1 | grep -v '^\.$' | grep -v '^\.\.$' > $assert_file || true
+  --let $include_silent = 1
+  --let $assert_count = 1
+  --let $assert_select = -.*-.*-.*-
+  --source include/assert_grep.inc
+  --enable_result_log
+  --let $include_silent = 0
+  --remove_file $assert_file
+}
+--enable_result_log
+
+--sleep 25
+
+--echo
+--echo *** Step 3: Check binlog size after read_timeout flush (should have increased)
+--let $size_file_after_first_timeout = $MYSQL_TMP_DIR/storage_size_after_first_timeout.txt
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$test_binlog' > '$size_file_after_first_timeout'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$test_binlog --query ContentLength --output text > $size_file_after_first_timeout 2>/dev/null || echo 0 > $size_file_after_first_timeout
+}
+--let $read_from_file = $size_file_after_first_timeout
+--source include/read_file_to_var.inc
+--let $size_after_first_timeout = $result
+
+--remove_file $size_file_after_first_timeout
+
+--echo
+--echo *** Verifying that data was flushed to storage after first read_timeout (PS-10320 fix)
+--let $size_increased_after_first = `SELECT IF($size_after_first_timeout > $size_initial, 1, 0)`
+if ($size_increased_after_first != 1)
+{
+  --die storage object size should have increased after first read_timeout flush, but it did not
+}
+
+--echo
+--echo *** Step 4: Wait around idle_time and generate new data in the binary log. verify either in idle_time or upon reconnection data was not flushed to storage.
+--sleep 10
+INSERT INTO t2(val) VALUES('test_data_after_timeout');
+
+# To test update_rows_partial event with binlog_server
+SET @save_binlog_row_value_options = @@GLOBAL.binlog_row_value_options;
+SET GLOBAL binlog_row_value_options = PARTIAL_JSON;
+
+UPDATE t2 set val='update_test_data_after_timeout' where val='test_data_after_timeout';
+
+CREATE TABLE tj (j JSON);
+INSERT INTO tj VALUES('{"name":"john","age":30,"city":"blr"}');
+INSERT INTO tj VALUES('{"arr":[1,2,3],"details":{"salary":5000}}');
+UPDATE tj SET j = JSON_REPLACE(j, '$.city', 'mysore') WHERE JSON_EXTRACT(j, '$.name') = 'john';
+UPDATE tj SET j = JSON_REPLACE(j, '$.arr[1]', 999);
+DROP TABLE tj;
+SET GLOBAL binlog_row_value_options = @save_binlog_row_value_options;
+
+--echo
+--echo *** Step 5: Check binlog size after around idle time (should NOT have changed - since no flush)
+--let $size_file_after_reconnection = $MYSQL_TMP_DIR/storage_size_after_reconnection.txt
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$test_binlog' > '$size_file_after_reconnection'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$test_binlog --query ContentLength --output text > $size_file_after_reconnection 2>/dev/null || echo 0 > $size_file_after_reconnection
+}
+--let $read_from_file = $size_file_after_reconnection
+--source include/read_file_to_var.inc
+--let $size_after_reconnection = $result
+--remove_file $size_file_after_reconnection
+
+--echo
+--echo *** Verifying that size did NOT change after reconnection (no flush)
+--let $size_unchanged = `SELECT IF($size_after_reconnection = $size_after_first_timeout, 1, 0)`
+if ($size_unchanged != 1)
+{
+  --die storage object size should NOT have changed after reconnection (no new data), but it did
+}
+
+--echo
+--echo *** Step 6: Execute INSERT again
+INSERT INTO t2(val) VALUES('test_data_after_timeout');
+
+--echo
+--echo *** Step 7: Check size immediately after INSERT ie around idle/reconnection time (should NOT have changed - data in buffer)
+--let $size_file_immediate = $MYSQL_TMP_DIR/storage_size_immediate.txt
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$test_binlog' > '$size_file_immediate'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$test_binlog --query ContentLength --output text > $size_file_immediate 2>/dev/null || echo 0 > $size_file_immediate
+}
+--let $read_from_file = $size_file_immediate
+--source include/read_file_to_var.inc
+--let $size_immediate = $result
+--remove_file $size_file_immediate
+
+--echo
+--echo *** Verifying that size did NOT change immediately after INSERT (data still in buffer)
+--let $size_unchanged_immediate = `SELECT IF($size_immediate = $size_after_reconnection, 1, 0)`
+if ($size_unchanged_immediate != 1)
+{
+  --die storage object size should NOT have changed immediately after INSERT (data in buffer), but it did
+}
+
+--echo
+--echo *** Step 8: Wait for read_timeout to trigger flush again (read_timeout=20)
+--sleep 35
+
+--echo
+--echo *** Step 9: Check binlog size after second read_timeout flush (should have increased)
+--let $size_file_after_second_timeout = $MYSQL_TMP_DIR/storage_size_after_second_timeout.txt
+if ($storage_backend == file)
+{
+  --exec /bin/sh -c "wc -c < '$binsrv_storage_path/$test_binlog' > '$size_file_after_second_timeout'"
+}
+if ($storage_backend == s3)
+{
+  --exec $aws_cli s3api head-object --bucket $MTR_BINSRV_AWS_S3_BUCKET --key $binsrv_storage_path/$test_binlog --query ContentLength --output text > $size_file_after_second_timeout 2>/dev/null || echo 0 > $size_file_after_second_timeout
+}
+--let $read_from_file = $size_file_after_second_timeout
+--source include/read_file_to_var.inc
+--let $size_after_second_timeout = $result
+--remove_file $size_file_after_second_timeout
+
+--echo
+--echo *** Verifying that size increased after second read_timeout flush
+--let $size_increased_after_second = `SELECT IF($size_after_second_timeout > $size_immediate, 1, 0)`
+if ($size_increased_after_second != 1)
+{
+  --die storage object size should have increased after second read_timeout flush, but it did not
+}
+
+FLUSH BINARY LOGS;
+
+--echo
+--echo *** Terminating the utility
+--source ../include/terminate_binsrv.inc
+
+--remove_file $binsrv_pid_file
+
+--disable_result_log
+if ($storage_backend == s3)
+{
+  --let $assert_text =  Temporary files shall be cleaned up in buffer directory (UUID format)
+  --let $assert_file = $MYSQL_TMP_DIR/buffer_check.log
+  --let $include_silent = 1
+  --disable_result_log
+  --exec ls -1 $binsrv_buffer_path 2>&1 | grep -v '^\.$' | grep -v '^\.\.$' > $assert_file || true
+  --let $assert_count = 0
+  --let $assert_select = -.*-.*-.*-
+  --source include/assert_grep.inc
+  --enable_result_log
+  --let $include_silent = 0
+  --remove_file $assert_file
+}
+--enable_result_log
+
+--exec $BINSRV fetch $binsrv_config_file_path > /dev/null
+
+--echo
+--echo *** Final verification: Compare storage object with server binlog
+--let $local_file = $binlog_base_dir/$test_binlog
+--let $storage_object = $binsrv_storage_path/$test_binlog
+--source ../include/diff_with_storage_object.inc
+
+--source ../include/tear_down_binsrv_environment.inc
+
+DROP TABLE t2;
+
+# Clean up any remaining binlog dump connections
+--let $binlog_dump_connection_id = `SELECT ID FROM performance_schema.processlist WHERE COMMAND = 'Binlog Dump'`
+if ($binlog_dump_connection_id != '')
+{
+  --disable_query_log
+  --replace_result $binlog_dump_connection_id <CONNECTION_ID>
+  eval KILL CONNECTION $binlog_dump_connection_id;
+  --enable_query_log
+}
+
+--source include/wait_until_count_sessions.inc