Весенняя партия Кафка Кафка в базу данных Работа - PullRequest
Новая
       34

Весенняя партия Кафка Кафка в базу данных Работа

2 голосов
/ 21 июня 2019

Мне нужен SpringReatch ItemReader для приема сообщений Kafka, результаты которых будут обработаны и записаны дальше.

Вот читатель предметов, который я реализовал: 

public abstract class KafkaItemReader<T> implements ItemReader<List<T>> {
  public abstract KafkaConsumer<String, T> getKafkaConsumer();

  public abstract String getTopic();

  public abstract long getPollingTime();

  @Override
  public List<T> read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
    Iterator<ConsumerRecord<String, T>> iterator = getKafkaConsumer()
        .poll(Duration.ofMillis(getPollingTime()))
        .records(getTopic())
        .iterator();
    List<T> records = new ArrayList<>();
    while (iterator.hasNext()) {
      records.add(iterator.next().value());
    }
    return records;
  }
}

Это следующие бобы для весеннего пакетного задания и шага: 

  @Bean
  public ItemWriter<List<DbEntity>> databaseWriter(DataSource dataSource) {
    //some item writer that needs to be implmented
    return null;
  }


  @Bean
  public Step kafkaToDatabaseStep(KafkaItemReader kafkaItemReader, //implementation of KafkaItemReader
                                  StepBuilderFactory stepBuilderFactory,
                                  DataSource dataSource) {

    return stepBuilderFactory
        .get("kafkaToDatabaseStep")
        .allowStartIfComplete(true)
        .<List<KafkaRecord>, List<DbEntity>>chunk(100)
        .reader(kafkaItemReader)
        .processor(itemProcessor()) //List<KafkaRecord> to List<DbEntity> converter
        .writer(databaseWriter(dataSource))
        .build();
  }


  @Bean
  public Job kafkaToDatabaseJob(
      @Qualifier("kafkaToDatabaseStep") Step step) {
    return jobBuilderFactory.get("kafkaToDatabaseJob")
        .incrementer(new RunIdIncrementer())
        .flow(step)
        .end()
        .build();
  }

Здесь я не знаю:

  1. Как зафиксировать смещение прочитанных сообщений в модуле записи, поскольку я хочу зафиксировать только после полной обработки записи.
  2. Как использовать JdbcBatchItemWriter в качестве ItemWriter в моем сценарии.

1 Ответ

0 голосов
/ 21 июня 2019

Предстоящая версия Spring Batch v4.2 GA обеспечит поддержку чтения / записи данных в разделах Apache Kafka.Вы уже можете попробовать это с 4.2.0.M2 релизом .

. Вы также можете взглянуть на Spring Tips об этом Джоша Лонга.

...