RetryHandler<String, String, String> handler  = (group, key, s) -> {
    CompletableFuture<String> future = new CompletableFuture<>();
    // 可以启动线程执行重试逻辑
    ...
    // 执行结束设置结果
    future.complete("retry");
    // future.completeExceptionally(new Throwable("err"));
    return future;
};

Retry<String, String, String> retry = new Retry<>(new MemoryStorage(new ConcurrentHashMap<>(), ArrayDeque::new)), handler);
retry.start();
...
// 加入重试队列
retry.push("topic1" , "some data");


// 应用关闭时
retry.shutdown();

RocksDBStorage db = new RocksDBStorage("any path");
db.open();

// 应用关闭时
db.close();

KafkaTemplate<String, String>                     template;

DefaultKafkaRetryHandler<String, String>          handler  = new DefaultKafkaRetryHandler<>(template);
Retry<String, String, SendResult<String, String>> retry   = new Retry<>(handler);
...