kafka消费端消费失败后怎么做后续处理？

@KafkaListener(topics = {"${kafka.topic.topicB}"}, groupId = "groupB")
    public void consumeTopicB(ConsumerRecord<String, String> consumerRecord, Acknowledgment acknowledgment) {
      

        Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());

        if(kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get(); 

            /*
             * 执行消费逻辑处理的代码，
             *
             */
            
          acknowledgment.acknowledge();// 消费成功后手动提交offset

            logger.info("消费者B消费topicB:{} partition:{}的消息 -> {}", consumerRecord.topic(), consumerRecord.partition(),message);

        }

public class KafkaMsg  implements Serializable {

    private static final long serialVersionUID = -1532915942422600087L;

    private String msgId;
    private  String  content;
    private  Integer  retryTime;  // 重试次数记录
    public String getMsgId() {
        return msgId;
    }

    public String getContent() {
        return content;
    }

    public void setMsgId(String msgId) {
        this.msgId = msgId;
    }

    public void setContent(String content) {
        this.content = content;
    }

     public Integer getRetryTime() {
        return retryTime;
    }

     public Integer setRetryTime(Integer time) {
         this.retryTime = time;
    }

    @Override
    public String toString() {
        return "KafkaMsg{" +
                "msgId='" + msgId + '\'' +
                ", content='" + content + '\'' +
                '}';
    }
}