
**Describe the unexpected behaviour**
After upgraded to ClickHouse 22.1.3 revision 54455,when using distributed table to insert data,errors occured in log.

**How to reproduce**
Version:ClickHouse 22.1.3.7

**Expected behavior**
After upgraded to ClickHouse 22.1.3 revision 54455,when using distributed table to insert data,errors occured in log.
Cluster's ClickHouse version are the same,22.1.3.7.

**Error message and/or stacktrace**
2022.01.28 18:39:41.098174 [ 80700 ] {} <Error> XXX.YYYY.DirectoryMonitor: Code: 246. DB::Exception: Unknown serialization kind 99: While sending /clickhouse/store/4db/4db5491f-555c-4c9d-8db5-491f555cbc9d/shard7_replica1/1571210.bin. (CORRUPTED_DATA), Stack trace (when copying this message, always include the lines below):

0. DB::Exception::Exception(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, int, bool) @ 0xa82d07a in /usr/bin/clickhouse
1. DB::SerializationInfo::deserializeFromKindsBinary(DB::ReadBuffer&) @ 0x1326c09c in /usr/bin/clickhouse
2. DB::NativeReader::read() @ 0x14826428 in /usr/bin/clickhouse
3. ? @ 0x14195913 in /usr/bin/clickhouse
4. DB::StorageDistributedDirectoryMonitor::processFile(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&) @ 0x1419391a in /usr/bin/clickhouse
5. DB::StorageDistributedDirectoryMonitor::run() @ 0x1418fa78 in /usr/bin/clickhouse
6. DB::BackgroundSchedulePoolTaskInfo::execute() @ 0x12f8470e in /usr/bin/clickhouse
7. DB::BackgroundSchedulePool::threadFunction() @ 0x12f870a7 in /usr/bin/clickhouse
8. ? @ 0x12f88170 in /usr/bin/clickhouse
9. ThreadPoolImpl<std::__1::thread>::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0xa86f4b7 in /usr/bin/clickhouse
10. ? @ 0xa872ebd in /usr/bin/clickhouse
11. start_thread @ 0x7ea5 in /usr/lib64/libpthread-2.17.so
12. __clone @ 0xfeb0d in /usr/lib64/libc-2.17.so
 (version 22.1.3.7 (official build))


{
    const auto pool_factory = [&storage, &name] (const Cluster::Address & address) -> ConnectionPoolPtr
    {
        const auto & cluster = storage.getCluster();
        const auto & shards_info = cluster->getShardsInfo();
        const auto & shards_addresses = cluster->getShardsAddresses();
        
        
        if (address.shard_index != 0)
        {
            if (!address.replica_index)
                throw Exception(ErrorCodes::INCORRECT_FILE_NAME,
                    "Wrong replica_index={} ({})", address.replica_index, name);
            if (address.shard_index > shards_info.size())
                throw Exception(ErrorCodes::INCORRECT_FILE_NAME,
                    "No shard with shard_index={} ({})", address.shard_index, name);
            const auto & shard_info = shards_info[address.shard_index - 1];
            if (address.replica_index > shard_info.per_replica_pools.size())
                throw Exception(ErrorCodes::INCORRECT_FILE_NAME,
                    "No shard with replica_index={} ({})", address.replica_index, name);
            return shard_info.per_replica_pools[address.replica_index - 1];
        }
        
        for (size_t shard_index = 0; shard_index < shards_info.size(); ++shard_index)
        {
            const Cluster::Addresses & replicas_addresses = shards_addresses[shard_index];
            for (size_t replica_index = 0; replica_index < replicas_addresses.size(); ++replica_index)