docs: public static <T> DataSet<Tuple2<Long, T>> zipWithIndex(DataSet<T> input) {
  
  		DataSet<Tuple2<Integer, Long>> elementCount = countElementsPerPartition(input);
  
  		return input.mapPartition(new RichMapPartitionFunction<T, Tuple2<Long, T>>() {
  
  			long start = 0;
  
  			@Override
  			public void open(Configuration parameters) throws Exception {
  				super.open(parameters);
  
  				List<Tuple2<Integer, Long>> offsets = getRuntimeContext().getBroadcastVariableWithInitializer(
  						"counts",
  						new BroadcastVariableInitializer<Tuple2<Integer, Long>, List<Tuple2<Integer, Long>>>() {
  							@Override
  							public List<Tuple2<Integer, Long>> initializeBroadcastVariable(Iterable<Tuple2<Integer, Long>> data) {
  								// sort the list by task id to calculate the correct offset
  								List<Tuple2<Integer, Long>> sortedData = new ArrayList<>();
  								for (Tuple2<Integer, Long> datum : data) {
  									sortedData.add(datum);
  								}
  								Collections.sort(sortedData, new Comparator<Tuple2<Integer, Long>>() {
  									@Override
  									public int compare(Tuple2<Integer, Long> o1, Tuple2<Integer, Long> o2) {
  										return o1.f0.compareTo(o2.f0);
  									}
  								});
  								return sortedData;
  							}
  						});
  
  				// compute the offset for each partition
  				for (int i = 0; i < getRuntimeContext().getIndexOfThisSubtask(); i++) {
  					start += offsets.get(i).f1;
  				}
  			}
  
  			@Override
  			public void mapPartition(Iterable<T> values, Collector<Tuple2<Long, T>> out) throws Exception {
  				for (T value: values) {
  					out.collect(new Tuple2<>(start++, value));
  				}
  			}
  		}).withBroadcastSet(elementCount, "counts");
  	}