StreamsApp.java

class StreamsApp {
    // properties, methods, etc
    
    public void forceInitializationOfStateStore(Map<String, Object> config, String sourceTopic, Deserializer keyDeserializer, Deserializer valueDeserializer) {
          config.put(ENABLE_AUTO_COMMIT_CONFIG, true);
          KafkaConsumer<String, String> consumer = new KafkaConsumer<>(config, keyDeserializer, valueDeserializer);
          ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
              public void onPartitionsRevoked (Collection< TopicPartition > partitions) {}
              public void onPartitionsAssigned (Collection < TopicPartition > partitions) {
                  List<TopicPartition> unInitialized = new ArrayList<>();
                  for (TopicPartition tp : partitions) {
                      OffsetAndMetadata offsetAndMetaData = consumer.committed(tp);
                      if (offsetAndMetaData == null) {
                          unInitialized.add(tp);
                      }
                  }
                  if (unInitialized.size() > 0) {
                      log.info("Forcing initialization of {} state stores for topic: {}", unInitialized.size(), sourceTopic);
                      consumer.seekToBeginning(unInitialized);
                  } else {
                      log.info("{} state stores have already been initialized for topic: {}", partitions.size(), sourceTopic);
                  }
              }
          };
          consumer.subscribe(Collections.singletonList(sourceTopic), listener);
          consumer.poll(1000L);
          consumer.close();
      }
  }