diff --git a/mysql-test/suite/rpl/r/rpl_timezone.result b/mysql-test/suite/rpl/r/rpl_timezone.result
index 927756dd530..f5e05bcb759 100644
--- a/mysql-test/suite/rpl/r/rpl_timezone.result
+++ b/mysql-test/suite/rpl/r/rpl_timezone.result
@@ -122,4 +122,22 @@ a	b
 SET @@session.time_zone = default;
 DROP TABLE t1;
 SET @@session.time_zone = default;
+CREATE TABLE t1 (date timestamp NOT NULL default CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP, a int(11) default NULL);
+SET @@session.time_zone='+01:00';
+insert into t1 values('2008-12-23 19:39:39',1);
+SET @@session.time_zone='+02:00';
+insert delayed into t1 values ('2008-12-23 19:39:39',2);
+flush table t1;
+flush logs;
+select * from t1;
+date	a
+2008-12-23 20:39:39	1
+2008-12-23 19:39:39	2
+DROP TABLE t1;
+select * from t1 order by a;
+date	a
+2008-12-23 20:39:39	1
+2008-12-23 19:39:39	2
+DROP TABLE t1;
+SET @@session.time_zone = default;
 End of 5.0 tests
diff --git a/mysql-test/suite/rpl/t/rpl_timezone.test b/mysql-test/suite/rpl/t/rpl_timezone.test
index d65242f2a22..85f60ae01cf 100644
--- a/mysql-test/suite/rpl/t/rpl_timezone.test
+++ b/mysql-test/suite/rpl/t/rpl_timezone.test
@@ -165,5 +165,31 @@ connection master;
 DROP TABLE t1;
 SET @@session.time_zone = default;
 
+# Bug#41719 delayed INSERT into timestamp col needs set time_zone for concurrent binlogging
+# To test that time_zone is correctly binloging for 'insert delayed' statement
+# Insert 2 values into timestamp col with different time_zone. Check result.
+
+--connection master
+CREATE TABLE t1 (date timestamp NOT NULL default CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP, a int(11) default NULL);
+
+SET @@session.time_zone='+01:00';
+insert into t1 values('2008-12-23 19:39:39',1);
+
+--connection master1
+SET @@session.time_zone='+02:00';
+insert delayed into t1 values ('2008-12-23 19:39:39',2);
+# Forces table t1 to be closed and flushes the query cache.
+# This makes sure that 'delayed insert' is executed before next statement.
+flush table t1;
+flush logs;
+select * from t1;
+DROP TABLE t1;
+
+let $MYSQLD_DATADIR= `select @@datadir;`;
+--exec $MYSQL_BINLOG $MYSQLD_DATADIR/master-bin.000001 | $MYSQL
+--connection master1
+select * from t1 order by a;
+DROP TABLE t1;
+SET @@session.time_zone = default;
 
 --echo End of 5.0 tests
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 0ea3824e46c..97eab6d2202 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1654,10 +1654,11 @@ public:
   ulong auto_increment_offset;
   timestamp_auto_set_type timestamp_field_type;
   LEX_STRING query;
+  Time_zone *time_zone;
 
   delayed_row(LEX_STRING const query_arg, enum_duplicates dup_arg,
               bool ignore_arg, bool log_query_arg)
-    : record(0), dup(dup_arg), ignore(ignore_arg), log_query(log_query_arg),
+    : record(0), time_zone(0), dup(dup_arg), ignore(ignore_arg), log_query(log_query_arg),
       forced_insert_id(0), query(query_arg)
     {}
   ~delayed_row()
@@ -2147,6 +2148,19 @@ int write_delayed(THD *thd, TABLE *table, enum_duplicates duplic,
     thd->first_successful_insert_id_in_prev_stmt;
   row->timestamp_field_type=    table->timestamp_field_type;
 
+  /* Add session variable timezone
+     Time_zone object will not be freed even the thread is ended.
+     So we can get time_zone object from thread which handling delayed statement.
+     See the comment of my_tz_find() for detail.
+  */
+  if (thd->time_zone_used)
+  {
+    row->time_zone = thd->variables.time_zone;
+  }
+  else
+  {
+    row->time_zone = NULL;
+  }
   /* Copy session variables. */
   row->auto_increment_increment= thd->variables.auto_increment_increment;
   row->auto_increment_offset=    thd->variables.auto_increment_offset;
@@ -2645,6 +2659,14 @@ bool Delayed_insert::handle_inserts(void)
 
     if (log_query && mysql_bin_log.is_open())
     {
+      bool backup_time_zone_used = thd.time_zone_used;
+      Time_zone *backup_time_zone = thd.variables.time_zone;
+      if (row->time_zone != NULL)
+      {
+        thd.time_zone_used = true;
+        thd.variables.time_zone = row->time_zone;
+      }
+
       /*
         If the query has several rows to insert, only the first row will come
         here. In row-based binlogging, this means that the first row will be
@@ -2656,6 +2678,9 @@ bool Delayed_insert::handle_inserts(void)
       thd.binlog_query(THD::ROW_QUERY_TYPE,
                        row->query.str, row->query.length,
                        FALSE, FALSE);
+
+      thd.time_zone_used = backup_time_zone_used;
+      thd.variables.time_zone = backup_time_zone;
     }
 
     if (table->s->blob_fields)