BUG#28647 Stop backup if disk full

storage/ndb/src/kernel/blocks/backup/Backup.cpp:
  Close files when disk full
storage/ndb/src/ndbapi/ndberror.c:
  Add error messages for backup failure
storage/ndb/test/src/NdbBackup.cpp:
  Add test case for disk full

mysql-test/r/bdb_notembedded.result

+set autocommit=1;
+reset master;
+create table bug16206 (a int);
+insert into bug16206 values(1);
+start transaction;
+insert into bug16206 values(2);
+commit;
+show binlog events;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+f	n	Format_desc	1	n	Server ver: VERSION, Binlog ver: 4
+f	n	Query	1	n	use `test`; create table bug16206 (a int)
+f	n	Query	1	n	use `test`; insert into bug16206 values(1)
+f	n	Query	1	n	use `test`; insert into bug16206 values(2)
+drop table bug16206;
+reset master;
+create table bug16206 (a int) engine=         bdb;
+insert into bug16206 values(0);
+insert into bug16206 values(1);
+start transaction;
+insert into bug16206 values(2);
+commit;
+insert into bug16206 values(3);
+show binlog events;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+f	n	Format_desc	1	n	Server ver: VERSION, Binlog ver: 4
+f	n	Query	1	n	use `test`; create table bug16206 (a int) engine=         bdb
+f	n	Query	1	n	use `test`; insert into bug16206 values(0)
+f	n	Query	1	n	use `test`; insert into bug16206 values(1)
+f	n	Query	1	n	use `test`; BEGIN
+f	n	Query	1	n	use `test`; insert into bug16206 values(2)
+f	n	Query	1	n	use `test`; COMMIT
+f	n	Query	1	n	use `test`; insert into bug16206 values(3)
+drop table bug16206;
+set autocommit=0;
+End of 5.0 tests

mysql-test/t/bdb_notembedded.test

+-- source include/not_embedded.inc
+-- source include/have_bdb.inc
+
+#
+# Bug #16206: Superfluous COMMIT event in binlog when updating BDB in autocommit mode
+#
+set autocommit=1;
+
+let $VERSION=`select version()`;
+
+reset master;
+create table bug16206 (a int);
+insert into bug16206 values(1);
+start transaction;
+insert into bug16206 values(2);
+commit;
+--replace_result $VERSION VERSION
+--replace_column 1 f 2 n 5 n
+show binlog events;
+drop table bug16206;
+
+reset master;
+create table bug16206 (a int) engine=         bdb;
+insert into bug16206 values(0);
+insert into bug16206 values(1);
+start transaction;
+insert into bug16206 values(2);
+commit;
+insert into bug16206 values(3);
+--replace_result $VERSION VERSION
+--replace_column 1 f 2 n 5 n
+show binlog events;
+drop table bug16206;
+
+set autocommit=0;
+
+
+--echo End of 5.0 tests

storage/ndb/src/kernel/blocks/backup/Backup.cpp

@@ -967,6 +967,7 @@ Backup::checkNodeFail(Signal* signal,
       ref->backupPtr = ptr.i;
       ref->backupId = ptr.p->backupId;
       ref->errorCode = AbortBackupOrd::BackupFailureDueToNodeFail;
+      ref->nodeId = getOwnNodeId();
       gsn= GSN_STOP_BACKUP_REF;
       len= StopBackupRef::SignalLength;
       pos= &ref->nodeId - signal->getDataPtr();
@@ -2080,6 +2081,15 @@ Backup::sendDropTrig(Signal* signal, BackupRecordPtr ptr)
     /**
      * Insert footers
      */
+    //if backup error, we needn't insert footers
+    if(ptr.p->checkError())
+    {
+      jam();
+      closeFiles(signal, ptr);
+      ptr.p->errorCode = 0;
+      return;
+    }
+
     {
       BackupFilePtr filePtr LINT_SET_PTR;
       ptr.p->files.getPtr(filePtr, ptr.p->logFilePtr);
@@ -4186,6 +4196,37 @@ Backup::checkFile(Signal* signal, BackupFilePtr filePtr)
 #if 0
   ndbout << "Ptr to data = " << hex << tmp << endl;
 #endif
+  BackupRecordPtr ptr LINT_SET_PTR;
+  c_backupPool.getPtr(ptr, filePtr.p->backupPtr);
+
+  if (ERROR_INSERTED(10036))
+  {
+    jam();
+    filePtr.p->m_flags &= ~(Uint32)BackupFile::BF_FILE_THREAD;
+    filePtr.p->errorCode = 2810;
+    ptr.p->setErrorCode(2810);
+
+    if(ptr.p->m_gsn == GSN_STOP_BACKUP_REQ)
+    {
+      jam();
+      closeFile(signal, ptr, filePtr);
+    }
+    return;
+  }
+
+  if(filePtr.p->errorCode != 0)
+  {
+    jam();
+    ptr.p->setErrorCode(filePtr.p->errorCode);
+
+    if(ptr.p->m_gsn == GSN_STOP_BACKUP_REQ)
+    {
+      jam();
+      closeFile(signal, ptr, filePtr);
+    }
+    return;
+  }
+
   if (!ready_to_write(ready, sz, eof, filePtr.p))
   {
     jam();
@@ -4217,8 +4258,6 @@ Backup::checkFile(Signal* signal, BackupFilePtr filePtr)
   ndbrequire(flags & BackupFile::BF_OPEN);
   ndbrequire(flags & BackupFile::BF_FILE_THREAD);
   
-  BackupRecordPtr ptr LINT_SET_PTR;
-  c_backupPool.getPtr(ptr, filePtr.p->backupPtr);
   closeFile(signal, ptr, filePtr);
 }
 
@@ -4581,6 +4620,22 @@ Backup::closeFilesDone(Signal* signal, BackupRecordPtr ptr)
   
   jam();
 
+  //error when do insert footer or close file
+  if(ptr.p->checkError())
+  {
+    StopBackupRef * ref = (StopBackupRef*)signal->getDataPtr();
+    ref->backupPtr = ptr.i;
+    ref->backupId = ptr.p->backupId;
+    ref->errorCode = ptr.p->errorCode;
+    ref->nodeId = getOwnNodeId();
+    sendSignal(ptr.p->masterRef, GSN_STOP_BACKUP_REF, signal,
+             StopBackupConf::SignalLength, JBB);
+
+    ptr.p->m_gsn = GSN_STOP_BACKUP_REF;
+    ptr.p->slaveState.setState(CLEANING);
+    return;
+  }
+
   StopBackupConf* conf = (StopBackupConf*)signal->getDataPtrSend();
   conf->backupId = ptr.p->backupId;
   conf->backupPtr = ptr.i;

storage/ndb/src/ndbapi/ndberror.c

@@ -624,6 +624,8 @@ ErrorBundle ErrorCodes[] = {
   { 4273, DMEC, IE, "No blob table in dict cache" },
   { 4274, DMEC, IE, "Corrupted main table PK in blob operation" },
   { 4275, DMEC, AE, "The blob method is incompatible with operation type or lock mode" },
+  { 2810, DMEC, TR, "No space left on the device" },
+  { 2815, DMEC, TR, "Error in reading files, please check file system" },
 
   { NO_CONTACT_WITH_PROCESS, DMEC, AE,
     "No contact with the process (dead ?)."},

storage/ndb/test/src/NdbBackup.cpp

@@ -350,7 +350,8 @@ FailS_codes[] = {
   10025,
   10027,
   10033,
-  10035
+  10035,
+  10036
 };
 
 int