testing recovering

Author: mageddo

src/main/java/com/mageddo/kafka/client/BatchConsumer.java

@@ -26,9 +26,9 @@ public void consume(
  commitFirstRecord(consumer, records, partition);
  }
  })
- .onExhausted(() -> {
+ .onExhausted((lastFailure) -> {
  log.info("status=exhausted-tries, records={}", records.count());
- records.forEach(record -> Consumers.doRecoverWhenAvailable(consumer, consumingConfig, record));
+ records.forEach(record -> Consumers.doRecoverWhenAvailable(consumer, consumingConfig, record, lastFailure));
  })
  .build();
 

src/main/java/com/mageddo/kafka/client/ConsumeHandler.java

@@ -1,5 +1,6 @@
 package com.mageddo.kafka.client;
 
+import lombok.AccessLevel;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -14,7 +15,7 @@
 import java.util.Map;
 
 @Data
-@Builder(toBuilder = true)
+@Builder(toBuilder = true, access = AccessLevel.PRIVATE)
 @Accessors(chain = true)
 @NoArgsConstructor
 @AllArgsConstructor
@@ -49,4 +50,7 @@ public ConsumerConfig<K, V> withProp(String k, Object v) {
  return this;
  }
 
+ public ConsumerConfig<K, V> copy(){
+ return this.toBuilder().build();
+ }
 }

src/main/java/com/mageddo/kafka/client/ConsumerConfig.java

@@ -24,10 +24,11 @@ public static <K, V> void commitSyncRecord(Consumer<K, V> consumer, ConsumerReco
  public static <K, V> void doRecoverWhenAvailable(
  Consumer<K, V> consumer,
  ConsumingConfig<K, V> consumingConfig,
- ConsumerRecord<K, V> record
+ ConsumerRecord<K, V> record,
+ Throwable lastFailure
  ) {
  if (consumingConfig.getRecoverCallback() != null) {
- consumingConfig.getRecoverCallback().recover(record);
+ consumingConfig.getRecoverCallback().recover(record, lastFailure);
  commitSyncRecord(consumer, record);
  } else {
  log.warn("status=no recover callback was specified");

src/main/java/com/mageddo/kafka/client/Consumers.java

@@ -21,9 +21,9 @@ public void consume(
  final AtomicBoolean recovered = new AtomicBoolean();
  Retrier
  .builder()
- .onExhausted(() -> {
+ .onExhausted((lastFailure) -> {
  log.info("exhausted tries");
- Consumers.doRecoverWhenAvailable(consumer, consumingConfig, record);
+ Consumers.doRecoverWhenAvailable(consumer, consumingConfig, record, lastFailure);
  recovered.set(true);
  })
  .onRetry(() -> {

src/main/java/com/mageddo/kafka/client/RecordConsumer.java

@@ -3,5 +3,5 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 public interface RecoverCallback<K, V> {
- void recover(ConsumerRecord<K, V> record);
+ void recover(ConsumerRecord<K, V> record, Throwable lastFailure);
 }

src/main/java/com/mageddo/kafka/client/RecoverCallback.java

@@ -1,6 +1,7 @@
 package com.mageddo.kafka.client;
 
 import java.util.ArrayList;
+import java.util.function.Consumer;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -19,7 +20,7 @@ public class Retrier {
 
  private RetryPolicy retryPolicy;
  private Callback onRetry;
- private Callback onExhausted;
+ private Consumer<Throwable> onExhausted;
 
  public void run(final Callback run) {
  Failsafe
@@ -28,7 +29,7 @@ public void run(final Callback run) {
  if(log.isTraceEnabled()){
  log.trace("status=onExhausted, ctx={}", ctx);
  }
- onExhausted.call();
+ onExhausted.accept(ctx.getLastFailure());
  return null;
  }),
  retryPolicyToFailSafeRetryPolicy(this.retryPolicy)

src/main/java/com/mageddo/kafka/client/Retrier.java

@@ -1,4 +1,4 @@
-import java.net.http.HttpClient;
+
 import java.time.Duration;
 import java.util.Collections;
 import java.util.LinkedHashMap;
@@ -64,13 +64,19 @@ public static void main(String[] args) {
  .withProp(VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName())
  .setRetryPolicy(RetryPolicy
  .builder()
- .maxTries(3)
+ .maxTries(1)
  .delay(Duration.ofSeconds(29))
  .build()
  )
+ .setRecoverCallback((record, lastFailure) -> {
+ log.info("status=recovering, record={}", new String(record.value()));
+ })
  .setBatchCallback((consumer, records, e) -> {
  for (final ConsumerRecord<String, byte[]> record : records) {
-// throw new RuntimeException("an error occurred");
+ final double randomValue = Math.random();
+ if(randomValue > 0.5 && randomValue < 0.8){
+ throw new RuntimeException("an error occurred");
+ }
  log.info("key={}, value={}", record.key(), new String(record.value()));
  }
  });