We use Kafka table engine to integrate with the Kafka. When we put the url into format_avro_schema_registry_url we can get the right message at the first time. 
`<format_avro_schema_registry_url>http://schema-registry-url:8081</format_avro_schema_registry_url>
`
However when there's any update in the schema-registry, we can't get message any more. There's some exceptions below: 
```
2020.11.12 02:46:36.694505 [ 126 ] {} <Error> void DB::StorageKafka::threadFunc(): Code: 1000, e.displayText() = DB::Exception: Timeout: connect timed out: 172.2.3.9:8081: while fetching schema id = 10, Stack trace (when copying this message, always include the lines below):
0. Poco::TimeoutException::TimeoutException(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, int) @ 0x1041d90f in /usr/bin/clickhouse
1. ? @ 0x102ffff5 in /usr/bin/clickhouse
2. Poco::Net::HTTPSession::connect(Poco::Net::SocketAddress const&) @ 0x102cccf5 in /usr/bin/clickhouse
3. Poco::Net::HTTPClientSession::reconnect() @ 0x102ba132 in /usr/bin/clickhouse
4. Poco::Net::HTTPClientSession::sendRequest(Poco::Net::HTTPRequest&) @ 0x102bafc0 in /usr/bin/clickhouse
5. DB::AvroConfluentRowInputFormat::SchemaRegistry::fetchSchema(unsigned int) @ 0xdb5ba91 in /usr/bin/clickhouse
6. std::__1::pair<std::__1::shared_ptr<avro::ValidSchema>, bool> DB::LRUCache<unsigned int, avro::ValidSchema, std::__1::hash<unsigned int>, DB::TrivialWeightFunction<avro::ValidSchema> >::getOrSet<DB::AvroConfluentRowInputFormat::SchemaRegistry::getSchema(unsigned int)::'lambda'()>(unsigned int const&, DB::AvroConfluentRowInputFormat::SchemaRegistry::getSchema(unsigned int)::'lambda'()&&) @ 0xdb5cdbb in /usr/bin/clickhouse
7. DB::AvroConfluentRowInputFormat::getOrCreateDeserializer(unsigned int) @ 0xdb42713 in /usr/bin/clickhouse
8. DB::AvroConfluentRowInputFormat::readRow(std::__1::vector<COW<DB::IColumn>::mutable_ptr<DB::IColumn>, std::__1::allocator<COW<DB::IColumn>::mutable_ptr<DB::IColumn> > >&, DB::RowReadExtension&) @ 0xdb42b4d in /usr/bin/clickhouse
9. DB::IRowInputFormat::generate() @ 0xdb5d151 in /usr/bin/clickhouse
10. DB::ISource::work() @ 0xdaf3d8b in /usr/bin/clickhouse
11. ? @ 0xd80106d in /usr/bin/clickhouse
12. DB::KafkaBlockInputStream::readImpl() @ 0xd801c48 in /usr/bin/clickhouse
13. DB::IBlockInputStream::read() @ 0xce4825d in /usr/bin/clickhouse
14. DB::copyData(DB::IBlockInputStream&, DB::IBlockOutputStream&, std::__1::atomic<bool>*) @ 0xce7717e in /usr/bin/clickhouse
15. DB::StorageKafka::streamToViews() @ 0xd7f0332 in /usr/bin/clickhouse
16. DB::StorageKafka::threadFunc() @ 0xd7f0da8 in /usr/bin/clickhouse
17. DB::BackgroundSchedulePoolTaskInfo::execute() @ 0xcfe9619 in /usr/bin/clickhouse
18. DB::BackgroundSchedulePool::threadFunction() @ 0xcfe9c42 in /usr/bin/clickhouse
19. ? @ 0xcfe9d72 in /usr/bin/clickhouse
20. ThreadPoolImpl<std::__1::thread>::worker(std::__1::__list_iterator<std::__1::thread, void*>) @ 0x902526b in /usr/bin/clickhouse
21. ? @ 0x9023753 in /usr/bin/clickhouse
22. start_thread @ 0x76db in /lib/x86_64-linux-gnu/libpthread-2.27.so
23. __clone @ 0x12188f in /lib/x86_64-linux-gnu/libc-2.27.so
 (version 20.4.4.18 (official build))
```
The timeout hostname 172.2.3.9:8081 is not the host of schema-registry-url:8081 at all. 

Does ClickHouse cache the DNS result of http://schema-registry-url:8081 at the first time and use the ip in the following calls?  When we use http://schema-registry-url:8081 in the ClickHouse server machine, we can get the right schema. 

How can we fix it? We already tried `System Drop DNS Cache` but it doesn't work. Since the schema update is very frequent in production environment, we don't want to restart the server every time the schema updates. 

{
    uint8_t magic;
    uint32_t schema_id;
    try
    {
        readBinaryBigEndian(magic, in);
        readBinaryBigEndian(schema_id, in);
    }
    catch (const Exception & e)
    {
        if (e.code() == ErrorCodes::CANNOT_READ_ALL_DATA)
        {
            
            throw Exception("Missing AvroConfluent magic byte or schema identifier.", ErrorCodes::INCORRECT_DATA);
        }
        else
            throw;
    }
    if (magic != 0x00)
    {
        throw Exception("Invalid magic byte before AvroConfluent schema identifier."
            " Must be zero byte, found " + std::to_string(int(magic)) + " instead", ErrorCodes::INCORRECT_DATA);
    }
    return schema_id;
}
AvroConfluentRowInputFormat::AvroConfluentRowInputFormat(