BUG#25256910: NODE RECOVERY FAILS WITH ERROR_CODE: 1062; HANDLER ERROR HA_ERR_FOUND_DUPP_KEY

On Single Primary Mode Group Replication certification process is disabled,
only one server accepts writes so there aren't conflicts. Exist one exception
for when a new primary server is being elected.

This causes a problem for when a new server is entering the group, it doesn't
know if certification is enable or not in the group as a consequence of the
primary election mechanism. Hence, queries being rejected in other member can
be accepted and fail on the new joiner.

This patch solves this problem sending on view change, when a member join or
leaves, the state of conflict detection of the members. When the member enters
it will receive the state of conflict detection of all members and activate it
if one of the members has conflict detection active.

Author: Anibal Pinto

rapid/plugin/group_replication/include/certifier.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -762,7 +762,7 @@ class Gtid_Executed_Message: public Plugin_gcs_message
    Implementation of the template methods of Gcs_plugin_message
    */
   void encode_payload(std::vector<unsigned char>* buffer) const;
-  void decode_payload(const unsigned char* buffer, uint64 length);
+  void decode_payload(const unsigned char* buffer, const unsigned char* end);
 
 private:
   std::vector<uchar> data;

rapid/plugin/group_replication/include/gcs_plugin_messages.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -245,9 +245,10 @@ class Plugin_gcs_message
     values according to the values decoded.
 
     @param[in] buffer the buffer to decode from.
-    @param[in] length the length of the buffer.
+    @param[in] end    the end of the buffer.
   */
-  virtual void decode_payload(const unsigned char* buffer, uint64 length)= 0;
+  virtual void decode_payload(const unsigned char* buffer,
+                              const unsigned char* end)= 0;
 
   /**
     Encodes the given payload item type and length into the buffer.

rapid/plugin/group_replication/include/member_info.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -97,8 +97,11 @@ class Group_member_info: public Plugin_gcs_message
     // length of the configuration flags: 4 bytes
     PIT_CONFIGURATION_FLAGS= 12,
 
+    // length of the conflict detection enabled: 1 byte
+    PIT_CONFLICT_DETECTION_ENABLE= 13,
+
     // No valid type codes can appear after this one.
-    PIT_MAX= 13
+    PIT_MAX= 14
   };
 
   /*
@@ -314,9 +317,24 @@ class Group_member_info: public Plugin_gcs_message
    */
   bool is_unreachable();
 
+  /**
+    Update this member conflict detection to true
+   */
+  void enable_conflict_detection();
+
+  /**
+    Update this member conflict detection to false
+   */
+  void disable_conflict_detection();
+
+  /**
+    Return true if conflict detection is enable on this member
+   */
+  bool is_conflict_detection_enabled();
+
 protected:
   void encode_payload(std::vector<unsigned char>* buffer) const;
-  void decode_payload(const unsigned char* buffer, uint64 length);
+  void decode_payload(const unsigned char* buffer, const unsigned char* end);
 
 private:
   std::string hostname;
@@ -332,6 +350,7 @@ class Group_member_info: public Plugin_gcs_message
   bool unreachable;
   Group_member_role role;
   uint32 configuration_flags;
+  bool conflict_detection_enable;
 };
 
 
@@ -446,6 +465,13 @@ class Group_member_info_manager_interface
    */
   virtual std::vector<Group_member_info*>* decode(const uchar* to_decode,
                                                   uint64 length)= 0;
+
+  /**¬
+  Check if some member of the group has the conflict detection enable
+
+  @return true if at least one member has  conflict detection enabled
+  */
+  virtual bool is_conflict_detection_enabled()= 0;
 };
 
 
@@ -492,6 +518,8 @@ class Group_member_info_manager: public Group_member_info_manager_interface
   std::vector<Group_member_info*>* decode(const uchar* to_decode,
                                           uint64 length);
 
+  bool is_conflict_detection_enabled();
+
 private:
   void clear_members();
 
@@ -580,7 +608,7 @@ class Group_member_info_manager_message: public Plugin_gcs_message
 
 protected:
   void encode_payload(std::vector<unsigned char>* buffer) const;
-  void decode_payload(const unsigned char* buffer, uint64 length);
+  void decode_payload(const unsigned char* buffer, const unsigned char* end);
 
 private:
   /**

rapid/plugin/group_replication/include/observer_trans.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2013, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2013, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -89,7 +89,7 @@ class Transaction_Message: public Plugin_gcs_message
    Implementation of the template methods
    */
   void encode_payload(std::vector<unsigned char>* buffer) const;
-  void decode_payload(const unsigned char* buffer, uint64 length);
+  void decode_payload(const unsigned char* buffer, const unsigned char* length);
 
 private:
   std::vector<uchar> data;

rapid/plugin/group_replication/include/pipeline_stats.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2016, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -151,9 +151,9 @@ class Pipeline_stats_member_message : public Plugin_gcs_message
     Message decoding method
 
     @param[in] buffer the received data
-    @param[in] length the received data size
+    @param[in] end    the end of the buffer
   */
-  void decode_payload(const unsigned char *buffer, uint64 length);
+  void decode_payload(const unsigned char *buffer, const unsigned char* end);
 
 private:
   int32 m_transactions_waiting_certification;

rapid/plugin/group_replication/include/recovery_message.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -99,9 +99,9 @@ class Recovery_message : public Plugin_gcs_message
     Message decoding method
 
     @param[in] buffer the received data
-    @param[in] length the received data size
+    @param[in] end    the end of the buffer
   */
-  void decode_payload(const unsigned char* buffer, uint64 length);
+  void decode_payload(const unsigned char* buffer, const unsigned char* end);
 
 private:
   /**The message type*/

rapid/plugin/group_replication/include/single_primary_message.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2016, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -89,9 +89,10 @@ class Single_primary_message : public Plugin_gcs_message
     Message decoding method
 
     @param[in] buffer the received data
-    @param[in] length the received data size
+    @param[in] end    the end of buffer
   */
-  void decode_payload(const unsigned char* buffer, uint64 length);
+  void decode_payload(const unsigned char* buffer,
+                      const unsigned char* end);
 
 private:
   /**The message type*/

rapid/plugin/group_replication/src/certifier.cc

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -1666,6 +1666,7 @@ void Certifier::enable_conflict_detection()
 
   mysql_mutex_lock(&LOCK_certification_info);
   conflict_detection_enable= true;
+  local_member_info->enable_conflict_detection();
   mysql_mutex_unlock(&LOCK_certification_info);
 
   log_message(MY_INFORMATION_LEVEL,
@@ -1682,6 +1683,7 @@ void Certifier::disable_conflict_detection()
 
   mysql_mutex_lock(&LOCK_certification_info);
   conflict_detection_enable= false;
+  local_member_info->disable_conflict_detection();
   mysql_mutex_unlock(&LOCK_certification_info);
 
   log_message(MY_INFORMATION_LEVEL,
@@ -1733,7 +1735,8 @@ Gtid_Executed_Message::encode_payload(std::vector<unsigned char>* buffer) const
 }
 
 void
-Gtid_Executed_Message::decode_payload(const unsigned char* buffer, uint64 length)
+Gtid_Executed_Message::decode_payload(const unsigned char* buffer,
+                                      const unsigned char* length)
 {
   DBUG_ENTER("Gtid_Executed_Message::decode_payload");
   const unsigned char *slider= buffer;

rapid/plugin/group_replication/src/gcs_event_handlers.cc

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -84,7 +84,7 @@ Plugin_gcs_events_handler::on_message_received(const Gcs_message& message) const
     break;
 
   default:
-    DBUG_ASSERT(0); /* purecov: inspected */
+    break; /* purecov: inspected */
   }
 }
 
@@ -332,6 +332,15 @@ Plugin_gcs_events_handler::on_view_changed(const Gcs_view& new_view,
   //update the Group Manager with all the received states
   this->update_group_info_manager(new_view, exchanged_data, is_leaving);
 
+  //enable conflict detection if someone on group have it enabled
+  if (local_member_info->in_primary_mode() &&
+      group_member_mgr->is_conflict_detection_enabled())
+  {
+    Certifier_interface *certifier=
+        this->applier_module->get_certification_handler()->get_certifier();
+    certifier->enable_conflict_detection();
+  }
+
   //Inform any interested handler that the view changed
   View_change_pipeline_action *vc_action=
     new View_change_pipeline_action(is_leaving);

rapid/plugin/group_replication/src/gcs_plugin_messages.cc

@@ -1,4 +1,4 @@
-/* Copyright (c) 2014, 2016, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2014, 2017, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -75,6 +75,7 @@ void Plugin_gcs_message::decode(const unsigned char* buffer,
 {
   DBUG_ENTER("Plugin_gcs_message::decode");
   const unsigned char *slider= buffer;
+  const unsigned char *end= buffer + length;
 
   m_version= uint4korr(slider);
   slider += WIRE_VERSION_SIZE;
@@ -92,7 +93,7 @@ void Plugin_gcs_message::decode(const unsigned char* buffer,
       s_cargo_type;
   slider += WIRE_CARGO_TYPE_SIZE;
 
-  decode_payload(slider, length);
+  decode_payload(slider, end);
 
   DBUG_VOID_RETURN;
 }