code: @Override
      public void ack() {
          waitClientRunning();
          if (!running) {
              return;
          }
  
          if (kafkaConsumer != null) {
              kafkaConsumer.commitSync();
          }
          if (kafkaConsumer2 != null) {
              kafkaConsumer2.commitSync();
          }
      }
docs: 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败