Skip to content

Latest commit

 

History

History
93 lines (79 loc) · 2.72 KB

README.md

File metadata and controls

93 lines (79 loc) · 2.72 KB
Raw

Rapids Kafka Client

Kafka Client is a vanilla java library that makes it easy to consume data from kafka, a list of features:

  • Parallel consuming
  • Consuming retry
  • Consuming failover
  • Designed to be easy to mock and test
  • Designed to support slow consumers without kafka re balancing
  • Designed to high throughput usage
  • Individual record consuming
  • Batch records consuming
  • Frameworkless, but easily configurable to some when wanted
  • Commits managed for you based on behavior
  • Low CPU usage

Getting Started

compile("com.mageddo.rapids-kafka-client:rapids-kafka-client:2.0.6")
ConsumerConfig.<String, String>builder()
.prop(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
.prop(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
.prop(GROUP_ID_CONFIG, "stocks")
.topics("stock_changed")
.recoverCallback(ctx -> {
  // here you can send the message to another topic, send a SMS, etc.
  log.info("status=recovering, value={}", ctx.record().value());
})
.callback((ctx, record) -> {
  log.info("status=consumed, value={}", record.value());
})
.build()
.consume()
.waitFor();

Making it easy to configure many consumers

public static void main(String[] args) {
  final ConsumerStarter consumerStarter = ConsumerStarter.start(defaultConfig(), Arrays.asList(
      new StockConsumer() // and many other consumers
  ));
  consumerStarter.waitFor();
  //    consumerStarter.stop();
}

static ConsumerConfig defaultConfig() {
  return ConsumerConfig.builder()
    .prop(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    .prop(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
    .prop(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
    .prop(GROUP_ID_CONFIG, "my-group-id")
    .build();
}

static class StockConsumer implements Consumer {

  ConsumeCallback<String, String> consume() {
    return (callbackContext, record) -> {
      System.out.printf("message from kafka: %s\n", record.value());
    };
  }

  @Override
  public ConsumerConfig<String, String> config() {
    return ConsumerConfig
      .<String, String>builder()
      .topics("stocks_events")
      .consumers(1)
      .callback(this.consume())
      .build();
  }
}

Examples