diff --git a/mysql-test/suite/innodb/r/dml_rollback_after_online_ddl_crash.result b/mysql-test/suite/innodb/r/dml_rollback_after_online_ddl_crash.result
new file mode 100644
index 00000000000..88f27fef9ef
--- /dev/null
+++ b/mysql-test/suite/innodb/r/dml_rollback_after_online_ddl_crash.result
@@ -0,0 +1,18 @@
+set global innodb_print_ddl_logs = ON;
+CREATE TABLE tt (a INT PRIMARY KEY, b INT);
+insert into tt values(1,1), (2,2), (3,3), (4,4);
+SET DEBUG_SYNC= 'alter_table_inplace_after_lock_downgrade SIGNAL downgrade WAIT_FOR go';
+alter table tt drop column b;;
+set session debug = "+d, crash_after_flush_engine_log";
+SET DEBUG_SYNC= 'now WAIT_FOR downgrade';
+# Crash right after flushing InnoDB redo log
+INSERT INTO tt VALUES(10,10);
+ERROR HY000: Lost connection to MySQL server during query
+# restart
+select * from tt;
+a	b
+1	1
+2	2
+3	3
+4	4
+drop table tt;
diff --git a/mysql-test/suite/innodb/t/dml_rollback_after_online_ddl_crash.test b/mysql-test/suite/innodb/t/dml_rollback_after_online_ddl_crash.test
new file mode 100644
index 00000000000..88790071e4c
--- /dev/null
+++ b/mysql-test/suite/innodb/t/dml_rollback_after_online_ddl_crash.test
@@ -0,0 +1,34 @@
+--source include/have_log_bin.inc
+
+connect(con1,localhost,root,,);
+connect(con2,localhost,root,,);
+
+set global innodb_print_ddl_logs = ON;
+
+connection con1;
+CREATE TABLE tt (a INT PRIMARY KEY, b INT);
+insert into tt values(1,1), (2,2), (3,3), (4,4);
+
+SET DEBUG_SYNC= 'alter_table_inplace_after_lock_downgrade SIGNAL downgrade WAIT_FOR go';
+--send alter table tt drop column b;
+
+connection con2;
+
+--let $_server_id=`SELECT @@server_id`
+--let $_expect_file_name=$MYSQLTEST_VARDIR/tmp/mysqld.$_server_id.expect
+--exec echo "restart" > $_expect_file_name
+
+set session debug = "+d, crash_after_flush_engine_log";
+SET DEBUG_SYNC= 'now WAIT_FOR downgrade';
+--echo # Crash right after flushing InnoDB redo log
+--error 2013
+INSERT INTO tt VALUES(10,10);
+
+--source include/start_mysqld.inc
+
+#connection default;
+select * from tt;
+drop table tt;
+
+disconnect con1;
+disconnect con2;
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index afadd244292..bb4f6f66995 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -94,8 +94,10 @@ trx_t *trx_allocate_for_mysql(void);
  @return own: transaction object */
 trx_t *trx_allocate_for_background(void);
 
-/** Resurrect table locks for resurrected transactions. */
-void trx_resurrect_locks();
+/** Resurrect table locks for resurrected transactions.
+@param[in]      all     false: resurrect locks for dictionary transactions,
+                        true : resurrect locks for all transcations. */
+void trx_resurrect_locks(bool all);
 
 /** Free and initialize a transaction object instantiated during recovery.
 @param[in,out]	trx	transaction object to free and initialize */
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index b5c08f6f651..73d20ed13ce 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -2882,7 +2882,8 @@ static void apply_dynamic_metadata() {
 /** On a restart, initialize the remaining InnoDB subsystems so that
 any tables (including data dictionary tables) can be accessed. */
 void srv_dict_recover_on_restart() {
-  trx_resurrect_locks();
+  /* Resurrect locks for dictionary transactions */
+  trx_resurrect_locks(false);
 
   /* Roll back any recovered data dictionary transactions, so
   that the data dictionary tables will be free of any locks.
@@ -2892,6 +2893,9 @@ void srv_dict_recover_on_restart() {
     trx_rollback_or_clean_recovered(FALSE);
   }
 
+  /* Resurrect locks for non-dictionary transactions */
+  trx_resurrect_locks(true);
+
   /* Do after all DD transactions recovery, to get consistent metadata */
   apply_dynamic_metadata();
 
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index f3ae03bdf9e..d4434c93c6a 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -727,13 +727,19 @@ static void trx_resurrect_table_ids(trx_t *trx, const trx_undo_ptr_t *undo_ptr,
   mtr_commit(&mtr);
 }
 
-/** Resurrect table locks for resurrected transactions. */
-void trx_resurrect_locks() {
+/** Resurrect table locks for resurrected transactions.
+@param[in]      all     false: resurrect locks for dictionary transactions,
+                        true : resurrect locks for all transcations. */
+void trx_resurrect_locks(bool all) {
   for (trx_table_map::const_iterator t = resurrected_trx_tables.begin();
        t != resurrected_trx_tables.end(); t++) {
     trx_t *trx = t->first;
     const table_id_set &tables = t->second;
-    ut_ad(trx->is_recovered);
+
+    /* If all is false, we skip non dictionary transactions. */
+    if (!trx->is_recovered || (!all && !trx->ddl_operation)) {
+      continue;
+    }
 
     for (table_id_set::const_iterator i = tables.begin(); i != tables.end();
          i++) {
@@ -765,7 +771,9 @@ void trx_resurrect_locks() {
     }
   }
 
-  resurrected_trx_tables.clear();
+  if (all) {
+    resurrected_trx_tables.clear();
+  }
 }
 
 /** Resurrect the transactions that were doing inserts at the time of the