Fix error when parse snapshot acl of v0

src/Service/NuRaftLogSnapshot.cpp

@@ -149,7 +149,7 @@ void KeeperSnapshotStore::serializeNode(
     }
 
     LOG_TRACE(log, "Append node path {}", path);
-    appendNodeToBatch(batch, path, node_copy);
+    appendNodeToBatch(batch, path, node_copy, version);
     processed++;
 
     String path_with_slash = path;
@@ -222,7 +222,7 @@ void KeeperSnapshotStore::serializeNodeV2(
     }
 
     LOG_TRACE(log, "Append node path {}", path);
-    appendNodeToBatchV2(batch, path, node_copy);
+    appendNodeToBatchV2(batch, path, node_copy, version);
     processed++;
 
     String path_with_slash = path;
@@ -233,14 +233,21 @@ void KeeperSnapshotStore::serializeNodeV2(
         serializeNodeV2(out, batch, store, path_with_slash + child, processed, checksum);
 }
 
-void KeeperSnapshotStore::appendNodeToBatch(ptr<SnapshotBatchPB> batch, const String & path, std::shared_ptr<KeeperNode> node)
+void KeeperSnapshotStore::appendNodeToBatch(ptr<SnapshotBatchPB> batch, const String & path, std::shared_ptr<KeeperNode> node, SnapshotVersion version)
 {
     SnapshotItemPB * entry = batch->add_data();
     WriteBufferFromNuraftBuffer buf;
 
     Coordination::write(path, buf);
     Coordination::write(node->data, buf);
-    Coordination::write(node->acl_id, buf);
+    if (version == SnapshotVersion::V0)
+    {
+        /// Just ignore acls for snapshot V0 /// TODO delete
+        Coordination::ACLs acls;
+        Coordination::write(acls, buf);
+    }
+    else
+        Coordination::write(node->acl_id, buf);
     Coordination::write(node->is_ephemeral, buf);
     Coordination::write(node->is_sequential, buf);
     Coordination::write(node->stat, buf);
@@ -250,13 +257,20 @@ void KeeperSnapshotStore::appendNodeToBatch(ptr<SnapshotBatchPB> batch, const St
     entry->set_data(String(reinterpret_cast<char *>(data->data_begin()), data->size()));
 }
 
-void KeeperSnapshotStore::appendNodeToBatchV2(ptr<SnapshotBatchBody> batch, const String & path, std::shared_ptr<KeeperNode> node)
+void KeeperSnapshotStore::appendNodeToBatchV2(ptr<SnapshotBatchBody> batch, const String & path, std::shared_ptr<KeeperNode> node, SnapshotVersion version)
 {
     WriteBufferFromNuraftBuffer buf;
 
     Coordination::write(path, buf);
     Coordination::write(node->data, buf);
-    Coordination::write(node->acl_id, buf);
+    if (version == SnapshotVersion::V0)
+    {
+        /// Just ignore acls for snapshot V0 /// TODO delete
+        Coordination::ACLs acls;
+        Coordination::write(acls, buf);
+    }
+    else
+        Coordination::write(node->acl_id, buf);
     Coordination::write(node->is_ephemeral, buf);
     Coordination::write(node->is_sequential, buf);
     Coordination::write(node->stat, buf);
@@ -608,7 +622,7 @@ bool KeeperSnapshotStore::parseBatchBody(KeeperStore & store, char * batch_buf,
 
                     if (ephemeral_owner != 0)
                     {
-                        LOG_TRACE(log, "Load snapshot find ephemeral node {} - {}", ephemeral_owner, key);
+                        LOG_TRACE(log, "Load snapshot find ephemeral node {} owner {}", key, ephemeral_owner);
                         std::lock_guard l(store.ephemerals_mutex);
                         auto & ephemeral_nodes = store.ephemerals[ephemeral_owner];
                         ephemeral_nodes.emplace(key);

src/Service/NuRaftLogSnapshot.h

@@ -140,9 +140,11 @@ class KeeperSnapshotStore
         uint32_t & checksum);
 
     /// Append node to batch
-    inline static void appendNodeToBatch(ptr<SnapshotBatchPB> batch, const String & path, std::shared_ptr<KeeperNode> node);
+    inline static void
+    appendNodeToBatch(ptr<SnapshotBatchPB> batch, const String & path, std::shared_ptr<KeeperNode> node, SnapshotVersion version);
     /// For snapshot version v3
-    inline static void appendNodeToBatchV2(ptr<SnapshotBatchBody> batch, const String & path, std::shared_ptr<KeeperNode> node);
+    inline static void
+    appendNodeToBatchV2(ptr<SnapshotBatchBody> batch, const String & path, std::shared_ptr<KeeperNode> node, SnapshotVersion version);
 
     /// Snapshot directory, note than the directory may contain more than one snapshot.
     String snap_dir;
@@ -192,7 +194,12 @@ class KeeperSnapshotManager
 
     ~KeeperSnapshotManager() = default;
 
-    size_t createSnapshot(snapshot & meta, KeeperStore & store, int64_t next_zxid = 0, int64_t next_session_id = 0, SnapshotVersion version = CURRENT_SNAPSHOT_VERSION);
+    size_t createSnapshot(
+        snapshot & meta,
+        KeeperStore & store,
+        int64_t next_zxid = 0,
+        int64_t next_session_id = 0,
+        SnapshotVersion version = CURRENT_SNAPSHOT_VERSION);
 
     /// save snapshot meta, invoked when we receive an snapshot from leader.
     bool receiveSnapshotMeta(snapshot & meta);

src/Service/SnapshotCommon.cpp

@@ -17,14 +17,6 @@
 namespace RK
 {
 
-namespace ErrorCodes
-{
-    extern const int CHECKSUM_DOESNT_MATCH;
-    extern const int CORRUPTED_DATA;
-    extern const int UNKNOWN_FORMAT_VERSION;
-    extern const int ILLEGAL_TYPE_OF_ARGUMENT;
-}
-
 using nuraft::cs_new;
 
 String toString(SnapshotVersion version)

src/Service/tests/gtest_raft_snapshot.cpp

@@ -412,7 +412,7 @@ void compareKeeperStore(KeeperStore & store, KeeperStore & new_store, bool compa
                 ASSERT_EQ(new_node->acl_id, it->second->acl_id);
             }
 
-            ASSERT_EQ(new_node->is_ephemeral, it->second->is_ephemeral);
+            ASSERT_EQ(new_node->is_ephemeral, it->second->is_ephemeral) << "Ephemeral not equals for path " << it->first;
             ASSERT_EQ(new_node->is_sequential, it->second->is_sequential);
             ASSERT_EQ(new_node->stat, it->second->stat);
             ASSERT_EQ(new_node->children, it->second->children);
@@ -552,7 +552,6 @@ void parseSnapshot(const SnapshotVersion version1, const SnapshotVersion version
     }
 
     for (int i = 0; i < 1024; i++)
-
     {
         String key = std::to_string(i);
         String value = "table_" + key;