Correctly handle unknown changelog versions and snapshot deleting

programs/keeper/keeper_config.xml

@@ -41,6 +41,7 @@
                 <min_session_timeout_ms>10000</min_session_timeout_ms>
                 <session_timeout_ms>100000</session_timeout_ms>
                 <raft_logs_level>information</raft_logs_level>
+                <compress_logs>false</compress_logs>
                 <!-- All settings listed in https://github.com/ClickHouse/ClickHouse/blob/master/src/Coordination/CoordinationSettings.h -->
             </coordination_settings>
 

src/Coordination/Changelog.cpp

@@ -516,7 +516,7 @@ public:
 
                 if (record.header.version > CURRENT_CHANGELOG_VERSION)
                     throw Exception(
-                        ErrorCodes::UNKNOWN_FORMAT_VERSION, "Unsupported changelog version {} on path {}", record.header.version, filepath);
+                        ErrorCodes::UNKNOWN_FORMAT_VERSION, "Unsupported changelog version {} on path {}", static_cast<uint8_t>(record.header.version), filepath);
 
                 /// Read data
                 if (record.header.blob_size != 0)
@@ -1480,4 +1480,9 @@ void Changelog::setRaftServer(const nuraft::ptr<nuraft::raft_server> & raft_serv
     raft_server = raft_server_;
 }
 
+bool Changelog::isInitialized() const
+{
+    return initialized;
+}
+
 }

src/Coordination/Changelog.h

@@ -153,6 +153,8 @@ public:
 
     void setRaftServer(const nuraft::ptr<nuraft::raft_server> & raft_server_);
 
+    bool isInitialized() const;
+
     /// Fsync log to disk
     ~Changelog();
 

src/Coordination/KeeperLogStore.cpp

@@ -127,7 +127,8 @@ void KeeperLogStore::shutdownChangelog()
 bool KeeperLogStore::flushChangelogAndShutdown()
 {
     std::lock_guard lock(changelog_lock);
-    changelog.flush();
+    if (changelog.isInitialized())
+        changelog.flush();
     changelog.shutdown();
     return true;
 }

src/Coordination/KeeperSnapshotManager.cpp

@@ -779,7 +779,7 @@ void KeeperSnapshotManager::removeSnapshot(uint64_t log_idx)
     if (itr == existing_snapshots.end())
         throw Exception(ErrorCodes::UNKNOWN_SNAPSHOT, "Unknown snapshot with log index {}", log_idx);
     const auto & [path, disk] = itr->second;
-    disk->removeFile(path);
+    disk->removeFileIfExists(path);
     existing_snapshots.erase(itr);
 }
 
@@ -809,8 +809,16 @@ SnapshotFileInfo KeeperSnapshotManager::serializeSnapshotToDisk(const KeeperStor
     disk->removeFile(tmp_snapshot_file_name);
 
     existing_snapshots.emplace(up_to_log_idx, SnapshotFileInfo{snapshot_file_name, disk});
-    removeOutdatedSnapshotsIfNeeded();
-    moveSnapshotsIfNeeded();
+
+    try
+    {
+        removeOutdatedSnapshotsIfNeeded();
+        moveSnapshotsIfNeeded();
+    }
+    catch (...)
+    {
+        tryLogCurrentException(log, "Failed to cleanup and/or move older snapshots");
+    }
 
     return {snapshot_file_name, disk};
 }

tests/config/config.d/keeper_port.xml

@@ -20,6 +20,8 @@
             <election_timeout_lower_bound_ms>0</election_timeout_lower_bound_ms>
             <election_timeout_upper_bound_ms>0</election_timeout_upper_bound_ms>
 
+            <compress_logs>0</compress_logs>
+
             <async_replication>1</async_replication>
         </coordination_settings>