In recent ClickHouse versions (21.3) Avro format for kafka does not work at all.
In ClickHouse 20.3 one Avro formatted kafka message can read, than stall forever.
Reportedly in ClickHouse prior 20.3 Avro formatted kafka messages work Ok while contain only single record
auto schema = json_body->getValue<std::string>("schema");
                LOG_TRACE((&Poco::Logger::get("AvroConfluentRowInputFormat")), "Successfully fetched schema id = {}\n{}", id, schema);
                return avro::compileJsonSchemaFromString(schema);
            }
            catch (const Exception &)
            {
                throw;
            }
            catch (const Poco::Exception & e)
            {
                throw Exception(Exception::CreateFromPocoTag{}, e);
            }
            catch (const avro::Exception & e)
            {
                throw Exception(e.what(), ErrorCodes::INCORRECT_DATA);
            }
        }
        catch (Exception & e)
        {
            e.addMessage("while fetching schema id = " + std::to_string(id));
            throw;
        }
    }
    Poco::URI base_url;
    LRUCache<uint32_t, avro::ValidSchema> schema_cache;
};
using ConfluentSchemaRegistry = AvroConfluentRowInputFormat::SchemaRegistry;
#define SCHEMA_REGISTRY_CACHE_MAX_SIZE 1000
static LRUCache<std::string, ConfluentSchemaRegistry>  schema_registry_cache(SCHEMA_REGISTRY_CACHE_MAX_SIZE);
static std::shared_ptr<ConfluentSchemaRegistry> getConfluentSchemaRegistry(const FormatSettings & format_settings)