Не удается запросить локальное государственное хранилище в приложении Kafka Streams - PullRequest
Новая
       41

Не удается запросить локальное государственное хранилище в приложении Kafka Streams

0 голосов
/ 05 июня 2018

Я создаю приложение kafka streams с помощью spring-kafka для группировки записей по ключу и применения некоторой бизнес-логики.Я следую конфигурации, указанной в spring-kafka-streams doc , но проблема в том, что когда я хочу получить значение из локального хранилища, я получаю следующую ошибку: 

org.apache.kafka.streams.errors.InvalidStateStoreException: The state store, user-data-response-count, may have migrated to another instance.
  at org.apache.kafka.streams.state.internals.QueryableStoreProvider.getStore(QueryableStoreProvider.java:60)
  at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1053)
  at com.umantis.management.service.UserDataManagementService.broadcastUserDataRequest(UserDataManagementService.java:121)

Вот моя KafkaStreamsConfiguration: 

@Configuration
@EnableConfigurationProperties(EventsKafkaProperties.class)
@EnableKafka
@EnableKafkaStreams
public class KafkaConfiguration {

@Value("${app.kafka.streams.application-id}")
private String applicationId;

// This contains both the bootstrap servers and the schema registry url
@Autowired
private EventsKafkaProperties eventsKafkaProperties;

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public StreamsConfig streamsConfig() {
    Map<String, Object> props = new HashMap<>();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId);
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.eventsKafkaProperties.getBrokers());
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, this.eventsKafkaProperties.getSchemaRegistryUrl());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    return new StreamsConfig(props);
}

@Bean
public KGroupedStream<String, UserDataResponse> responseKStream(StreamsBuilder streamsBuilder, TopicUtils topicUtils) {
    final Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url", this.eventsKafkaProperties.getSchemaRegistryUrl());

    final Serde<UserDataResponse> valueSpecificAvroSerde = new SpecificAvroSerde<>();
    valueSpecificAvroSerde.configure(serdeConfig, false);

    return streamsBuilder
            .stream("myTopic", Consumed.with(Serdes.String(), valueSpecificAvroSerde))
            .groupByKey();
}

А вот мой код обслуживания не работает на getKafkaStreams().store: 

@Slf4j
@Service
public class UserDataManagementService {

    private static final String RESPONSE_COUNT_STORE = "user-data-response-count";

    @Autowired
    private StreamsBuilderFactoryBean streamsBuilderFactory;

    public UserDataResponse broadcastUserDataRequest() {
        this.responseGroupStream.count(Materialized.as(RESPONSE_COUNT_STORE));

        if (!this.streamsBuilderFactory.isRunning()) {
            throw new KafkaStoreNotAvailableException();
        }

        // here we should have a single running kafka instance
        ReadOnlyKeyValueStore<String, Long> countStore =
                this.streamsBuilderFactory.getKafkaStreams().store(RESPONSE_COUNT_STORE, QueryableStoreTypes.keyValueStore());

        ...
    }

Контекст: я запускаю приложение на одном экземпляре ввесенний тест загрузки, и я гарантирую, что экземпляр kafka находится в рабочем состоянии.Я искал документацию от apache по этой проблеме, но мой случай не соответствует.

Может кто-нибудь указать мне, что я делаю неправильно, и возможное решение?

Я довольно новичок в Kafka Streams, поэтому любая помощь будет высоко оценена.

1 Ответ

0 голосов
/ 06 июня 2018

Хорошо, только что увидел, что я спрашиваю, запущена ли фабрика потоков, но я не спрашивал, действительно ли запущен экземпляр потоков kakfa.

Опрос streamsBuilderFactory.getKafkaStreams().state решил проблему.

...