diff --git a/mysql-test/suite/binlog_in_engine/recovery.test b/mysql-test/suite/binlog_in_engine/recovery.test
index 88db68d20ad..b90303b9b17 100644
--- a/mysql-test/suite/binlog_in_engine/recovery.test
+++ b/mysql-test/suite/binlog_in_engine/recovery.test
@@ -8,7 +8,6 @@
 RESET MASTER;
 
 CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
---let $gtid_pos= `SELECT @@last_gtid`
 INSERT INTO t1 VALUES (1);
 
 --let $no_checkpoint_flush= 1
diff --git a/mysql-test/suite/binlog_in_engine/recovery_large.opt b/mysql-test/suite/binlog_in_engine/recovery_large.opt
new file mode 100644
index 00000000000..915f0638bd2
--- /dev/null
+++ b/mysql-test/suite/binlog_in_engine/recovery_large.opt
@@ -0,0 +1 @@
+--loose-skip-stack-trace --skip-core-file --max-binlog-size=64K --innodb-log-file-size=64M
diff --git a/mysql-test/suite/binlog_in_engine/recovery_large.result b/mysql-test/suite/binlog_in_engine/recovery_large.result
new file mode 100644
index 00000000000..2b4dc5d205a
--- /dev/null
+++ b/mysql-test/suite/binlog_in_engine/recovery_large.result
@@ -0,0 +1,45 @@
+include/master-slave.inc
+[connection master]
+connection slave;
+include/stop_slave.inc
+SET GLOBAL gtid_slave_pos= '';
+connection master;
+RESET MASTER;
+DROP TABLE IF EXISTS t1, t2;
+Warnings:
+Note	1051	Unknown table 'test.t1,test.t2'
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+CREATE TABLE t2 (a INT PRIMARY KEY AUTO_INCREMENT, b INT, c VARCHAR(4000)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+
+# Flush all dirty pages from buffer pool
+SET @no_checkpoint_save_pct= @@GLOBAL.innodb_max_dirty_pages_pct;
+SET @no_checkpoint_save_pct_lwm= @@GLOBAL.innodb_max_dirty_pages_pct_lwm;
+SET GLOBAL innodb_max_dirty_pages_pct_lwm=0.0;
+SET GLOBAL innodb_max_dirty_pages_pct=0.0;
+SET GLOBAL innodb_max_dirty_pages_pct= @no_checkpoint_save_pct;
+SET GLOBAL innodb_max_dirty_pages_pct_lwm= @no_checkpoint_save_pct_lwm;
+
+CHECKSUM TABLE t1, t2;
+Table	Checksum
+test.t1	659068978
+test.t2	4017444020
+SET SESSION debug_dbug="+d,crash_dispatch_command_before";
+SELECT 1;
+Got one of the listed errors
+connection default;
+HULU
+BULU
+connection server_1;
+connection master;
+include/save_master_gtid.inc
+connection slave;
+include/start_slave.inc
+include/sync_with_master_gtid.inc
+CHECKSUM TABLE t1, t2;
+Table	Checksum
+test.t1	659068978
+test.t2	4017444020
+connection master;
+DROP TABLE t1, t2;
+include/rpl_end.inc
diff --git a/mysql-test/suite/binlog_in_engine/recovery_large.test b/mysql-test/suite/binlog_in_engine/recovery_large.test
new file mode 100644
index 00000000000..21b9432de1f
--- /dev/null
+++ b/mysql-test/suite/binlog_in_engine/recovery_large.test
@@ -0,0 +1,120 @@
+--source include/not_embedded.inc
+--source include/not_valgrind.inc
+--source include/have_debug.inc
+--source include/have_binlog_format_row.inc
+--source include/master-slave.inc
+--source include/have_innodb_binlog.inc
+
+--let $pre_count= 4
+--let $mid_count= 55
+
+--connection slave
+--source include/stop_slave.inc
+SET GLOBAL gtid_slave_pos= '';
+
+--connection master
+--let $datadir= `SELECT @@datadir`
+RESET MASTER;
+
+DROP TABLE IF EXISTS t1, t2;
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+CREATE TABLE t2 (a INT PRIMARY KEY AUTO_INCREMENT, b INT, c VARCHAR(4000)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+
+--disable_query_log
+--let $i= 0
+while ($i < $pre_count) {
+  eval INSERT INTO t1 VALUES (100+$i);
+  eval INSERT INTO t2(b, c) VALUES ($i, repeat(chr(65 + ($i MOD 27)), 3500));
+  inc $i;
+}
+--enable_query_log
+
+--let $no_checkpoint_flush= 1
+--let $no_checkpoint_kill= 1
+--source ../../suite/innodb/include/no_checkpoint_start.inc
+--let $file= query_get_value(SHOW MASTER STATUS, File, 1)
+--let $pos= query_get_value(SHOW MASTER STATUS, Position, 1)
+
+
+--disable_query_log
+--let $i= 0
+while ($i < $mid_count) {
+  eval INSERT INTO t1 VALUES (1000+$i);
+  eval INSERT INTO t2(b, c) VALUES ($i, repeat(chr(65 + ($i MOD 27)), 3500));
+  inc $i;
+}
+--enable_query_log
+
+CHECKSUM TABLE t1, t2;
+
+# Crash the server
+# Set $_expect_file_name for no_checkpoint_end.inc which uses start_mysqld.inc
+--let $_expect_file_name= $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+--write_file $_expect_file_name
+wait-recovery.test
+EOF
+
+SET SESSION debug_dbug="+d,crash_dispatch_command_before";
+--error 2006,2013
+SELECT 1;
+--source include/wait_until_disconnected.inc
+--connection default
+--source include/wait_until_disconnected.inc
+
+--source ../../suite/innodb/include/no_checkpoint_end.inc
+
+--echo HULU
+
+# Overwrite the binlog file past the last checkpoint
+# But only in the two most recent files, the prior ones are durably flushed
+# and will not be recovered.
+--let BINLOG_FILE= $datadir/$file
+--let OFFSET= $pos
+--let LEN=5000
+perl;
+my $orig_file= $ENV{BINLOG_FILE};
+my $orig_pos= $ENV{OFFSET};
+$orig_file =~ m/(.*binlog-)([0-9]{6})\.ibb/ or die "Unexpected file name $file";
+my ($base, $seq)= ($1, $2);
+
+for (;;) {
+  my $file= $base . sprintf("%06d", $seq) . ".ibb";
+  last unless -f $file;
+  my $file_plus_2= $base . sprintf("%06d", $seq+2) . ".ibb";
+  ++$seq;
+  next if -f $file_plus_2;
+
+  my $pos= $file eq $orig_file ? $orig_pos : 0;
+  open F, '+<', $file or die $!;
+  sysseek F, $pos, 0 or die $!;
+  my $x= chr(0) x $ENV{LEN};
+  syswrite F, $x, $ENV{LEN} or die $!;
+}
+EOF
+
+--echo BULU
+
+--append_file $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
+restart
+EOF
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+--connection server_1
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+--connection master
+--enable_reconnect
+--source include/wait_until_connected_again.inc
+
+--source include/save_master_gtid.inc
+
+--connection slave
+--source include/start_slave.inc
+--source include/sync_with_master_gtid.inc
+CHECKSUM TABLE t1, t2;
+
+--connection master
+DROP TABLE t1, t2;
+
+--source include/rpl_end.inc
diff --git a/storage/innobase/fsp/fsp_binlog.cc b/storage/innobase/fsp/fsp_binlog.cc
index b68b1c84a8f..8bd9ef672ac 100644
--- a/storage/innobase/fsp/fsp_binlog.cc
+++ b/storage/innobase/fsp/fsp_binlog.cc
@@ -309,9 +309,12 @@ fsp_binlog_page_fifo::flush_up_to(uint64_t file_no, uint32_t page_no)
 void
 fsp_binlog_page_fifo::do_fdatasync(uint64_t file_no)
 {
+  File fh;
   mysql_mutex_lock(&m_mutex);
+  if (file_no < first_file_no)
+    goto done;   /* Old files are already fully synced. */
   ut_a(file_no == first_file_no || file_no == first_file_no + 1);
-  File fh= fifos[file_no & 1].fh;
+  fh= fifos[file_no & 1].fh;
   if (fh != (File)-1)
   {
     while (flushing)
@@ -324,6 +327,7 @@ fsp_binlog_page_fifo::do_fdatasync(uint64_t file_no)
     flushing= false;
     pthread_cond_signal(&m_cond);
   }
+done:
   mysql_mutex_unlock(&m_mutex);
 }
 
diff --git a/storage/innobase/handler/innodb_binlog.cc b/storage/innobase/handler/innodb_binlog.cc
index a1dd9aa2181..16863a8a742 100644
--- a/storage/innobase/handler/innodb_binlog.cc
+++ b/storage/innobase/handler/innodb_binlog.cc
@@ -628,13 +628,13 @@ bool binlog_recovery::init_recovery(bool space_id, uint32_t page_no,
 
     if (res2 < 0 && !srv_force_recovery)
     {
-      sql_print_error("InnoDB: I/O error reading binlog file number " PRIu64,
+      sql_print_error("InnoDB: I/O error reading binlog file number %" PRIu64,
                       file_no2);
       return true;
     }
     if (res1 < 0 && !srv_force_recovery)
     {
-      sql_print_error("InnoDB: I/O error reading binlog file number " PRIu64,
+      sql_print_error("InnoDB: I/O error reading binlog file number %" PRIu64,
                       file_no1);
       return true;
     }
@@ -945,10 +945,17 @@ binlog_recovery::apply_redo(bool space_id, uint32_t page_no, uint16_t offset,
     skipping_partial_page= false;
   }
 
-  if (start_lsn < start_file_lsn)
+  if (skipping_early_lsn)
   {
-    if (skipping_early_lsn)
+    if (start_lsn < start_file_lsn || space_id != (cur_file_no & 1))
       return false;  /* Skip record for earlier file that's already durable. */
+    /* Now reset the current page to match the real starting point. */
+    cur_page_no= page_no;
+  }
+
+  if (UNIV_UNLIKELY(start_lsn < start_file_lsn))
+  {
+    ut_a(!skipping_early_lsn /* Was handled in condition above */);
     if (!srv_force_recovery)
     {
       sql_print_error("InnoDB: Unexpected LSN " LSN_PF " during recovery, "
@@ -971,7 +978,7 @@ binlog_recovery::apply_redo(bool space_id, uint32_t page_no, uint16_t offset,
            cur_phys_size >> srv_page_size_shift) &&
          !srv_force_recovery)
     {
-      sql_print_error("InnoDB: Missing recovery record at end of file_no="
+      sql_print_error("InnoDB: Missing recovery record at end of file_no=%"
                       PRIu64 ", LSN " LSN_PF, cur_file_no, start_lsn);
       return true;
     }
@@ -979,7 +986,7 @@ binlog_recovery::apply_redo(bool space_id, uint32_t page_no, uint16_t offset,
     /* Check that we recover from the start of the next file. */
     if ((page_no > 0 || offset > FIL_PAGE_DATA) && !srv_force_recovery)
     {
-      sql_print_error("InnoDB: Missing recovery record at start of file_no="
+      sql_print_error("InnoDB: Missing recovery record at start of file_no=%"
                       PRIu64 ", LSN " LSN_PF, cur_file_no+1, start_lsn);
       return true;
     }
@@ -993,7 +1000,7 @@ binlog_recovery::apply_redo(bool space_id, uint32_t page_no, uint16_t offset,
     if (cur_page_offset < srv_page_size - FIL_PAGE_DATA_END &&
         !srv_force_recovery)
     {
-      sql_print_error("InnoDB: Missing recovery record in file_no="
+      sql_print_error("InnoDB: Missing recovery record in file_no=%"
                       PRIu64 ", page_no=%u, LSN " LSN_PF,
                       cur_file_no, cur_page_no, start_lsn);
       return true;
@@ -1002,7 +1009,7 @@ binlog_recovery::apply_redo(bool space_id, uint32_t page_no, uint16_t offset,
     if ((page_no != cur_page_no + 1 || offset > FIL_PAGE_DATA) &&
         !srv_force_recovery)
     {
-      sql_print_error("InnoDB: Missing recovery record in file_no="
+      sql_print_error("InnoDB: Missing recovery record in file_no=%"
                       PRIu64 ", page_no=%u, LSN " LSN_PF,
                       cur_file_no, cur_page_no + 1, start_lsn);
       return true;
@@ -1016,7 +1023,7 @@ binlog_recovery::apply_redo(bool space_id, uint32_t page_no, uint16_t offset,
            offset > FIL_PAGE_DATA &&
            !srv_force_recovery)
   {
-      sql_print_error("InnoDB: Missing recovery record in file_no="
+      sql_print_error("InnoDB: Missing recovery record in file_no=%"
                       PRIu64 ", page_no=%u, LSN " LSN_PF,
                       cur_file_no, cur_page_no, start_lsn);
       return true;