Почему мне нужно создать Kafka Consumer для подключения к реестру схемы?

Question

Предыдущее примечание: я довольно новичок в Kafka.

Я пытаюсь получить все схемы из реестра схем, но я не могу сделать это только с помощью клиента реестра схем.Это работает, только если до этого я создаю экземпляр KafkaConsumer.

Не могу понять почему.Вот код (с потребителем на месте).

ConsumerConfig - это просто класс со всеми необходимыми конфигурациями.Включая URL-адрес реестра схем.

Consumer<String, String>  consumer = new KafkaConsumer<String, String>(ConsumerConfig.get());
CachedSchemaRegistryClient client = new CachedSchemaRegistryClient(ConsumerConfig.getSchemaRegistryURL(), 30);
Collection<String> listOfSubjects = client.getAllSubjects();
consumer.close();

Без потребителя я получаю:

io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException: Нет содержимого длякарта из-за конца ввода

С потребителем все отлично работает.Хотелось бы, чтобы кто-то пролил некоторый свет на то, почему это происходит, если я не вижу причин для того, чтобы мне нужно было подключаться к фактическому кластеру Kafka через потребителя, чтобы получить доступ к реестру схем, который находится на другой конечной точке.

Answer 1

Вам вообще не нужно создавать экземпляр KafkaConsumer.Оба абсолютно независимы.

Если вы просто хотите получить все объекты и схемы из SchemaRegistry, просто создайте экземпляр CachedSchemaRegistryClient и вызовите связанную операцию.

Вот рабочий пример:

 private final static Map<String, Schema> schemas = new ConcurrentHashMap<>();
 protected static SchemaRegistryClient schemaRegistryClient;

 public static void main(String[] args) {
       String registryUrl = "http://localhost:8081";
        try {
            schemaRegistryClient = new CachedSchemaRegistryClient(registryUrl, 30);
            System.out.println(schemaRegistryClient);
            Collection<String> subjects = schemaRegistryClient.getAllSubjects();
            System.out.println(subjects);
        } catch (Exception e){
            throw new RuntimeException(e);
        }
    }