docs: public void single(EtlEventData data) {
          Assert.notNull(data);
          String path = StagePathUtils.getExtractStage(data.getPipelineId(), data.getProcessId());
          data.setCurrNid(ArbitrateConfigUtils.getCurrentNid());
          // 序列化
          byte[] bytes = JsonUtils.marshalToByte(data, SerializerFeature.WriteClassName);
          try {
              zookeeper.create(path, bytes, CreateMode.PERSISTENT);
          } catch (ZkNoNodeException e) {
              // process节点不存在，出现了rollback/shutdown操作，直接忽略
              logger.warn("pipelineId[{}] extract ignore processId[{}] single by data:{}",
                          new Object[] { data.getPipelineId(), data.getProcessId(), data });
          } catch (ZkNodeExistsException e) {
              // process节点已存在，出现了ConnectionLoss retry操作
              logger.warn("pipelineId[{}] extract ignore processId[{}] single by data:{}",
                          new Object[] { data.getPipelineId(), data.getProcessId(), data });
          } catch (ZkInterruptedException e) {
              // ignore
          } catch (ZkException e) {
              throw new ArbitrateException("Extract_single", e.getMessage(), e);
          }
      }