eventually, we must replace this with the o.a.m.d.s.Sink class as a parameter
346849
public void generate( org.codehaus.doxia.sink.Sink sink, Locale locale )
        throws MavenReportException
    {
        generate( (Sink) sink, locale );

    }
928546
private Object bootstrapSinksFromDisk()
  {
    Preconditions.checkState(sinks.isEmpty(), "Already bootstrapped?!");

    final File baseDir = tuningConfig.getBasePersistDirectory();
    if (!baseDir.exists()) {
      return null;
    }

    final File[] files = baseDir.listFiles();
    if (files == null) {
      return null;
    }


    final Committed committed;
    File commitFile = null;
    try {
      commitLock.lock();
      commitFile = computeCommitFile();
      if (commitFile.exists()) {
        committed = objectMapper.readValue(commitFile, Committed.class);
      } else {
        committed = Committed.nil();
      }
    }
    catch (Exception e) {
      throw new ISE(e, "Failed to read commitFile: %s", commitFile);
    }
    finally {
      commitLock.unlock();
    }

    int rowsSoFar = 0;

    log.info("Loading sinks from[%s]: %s", baseDir, committed.getHydrants().keySet());

    for (File sinkDir : files) {
      final File identifierFile = new File(sinkDir, IDENTIFIER_FILE_NAME);
      if (!identifierFile.isFile()) {
        // No identifier in this sinkDir; it must not actually be a sink directory. Skip it.
        continue;
      }

      try {
        final SegmentIdWithShardSpec identifier = objectMapper.readValue(
            new File(sinkDir, "identifier.json"),
            SegmentIdWithShardSpec.class
        );

        final int committedHydrants = committed.getCommittedHydrants(identifier.toString());

        if (committedHydrants <= 0) {
          log.info("Removing uncommitted sink at [%s]", sinkDir);
          FileUtils.deleteDirectory(sinkDir);
          continue;
        }

        // To avoid reading and listing of "merged" dir and other special files
        final File[] sinkFiles = sinkDir.listFiles(
            (dir, fileName) -> !(Ints.tryParse(fileName) == null)
        );

        Arrays.sort(
            sinkFiles,
            (o1, o2) -> Ints.compare(Integer.parseInt(o1.getName()), Integer.parseInt(o2.getName()))
        );

        List<FireHydrant> hydrants = new ArrayList<>();
        for (File hydrantDir : sinkFiles) {
          final int hydrantNumber = Integer.parseInt(hydrantDir.getName());

          if (hydrantNumber >= committedHydrants) {
            log.info("Removing uncommitted segment at [%s]", hydrantDir);
            FileUtils.deleteDirectory(hydrantDir);
          } else {
            log.info("Loading previously persisted segment at [%s]", hydrantDir);
            if (hydrantNumber != hydrants.size()) {
              throw new ISE("Missing hydrant [%,d] in sinkDir [%s].", hydrants.size(), sinkDir);
            }

            hydrants.add(
                new FireHydrant(
                    new QueryableIndexSegment(indexIO.loadIndex(hydrantDir), identifier.asSegmentId()),
                    hydrantNumber
                )
            );
          }
        }

        // Make sure we loaded enough hydrants.
        if (committedHydrants != hydrants.size()) {
          throw new ISE("Missing hydrant [%,d] in sinkDir [%s].", hydrants.size(), sinkDir);
        }

        Sink currSink = new Sink(
            identifier.getInterval(),
            schema,
            identifier.getShardSpec(),
            identifier.getVersion(),
            tuningConfig.getMaxRowsInMemory(),
            maxBytesTuningConfig,
            tuningConfig.isReportParseExceptions(),
            null,
            hydrants
        );
        rowsSoFar += currSink.getNumRows();
        sinks.put(identifier, currSink);
        sinkTimeline.add(
            currSink.getInterval(),
            currSink.getVersion(),
            identifier.getShardSpec().createChunk(currSink)
        );

        segmentAnnouncer.announceSegment(currSink.getSegment());
      }
      catch (IOException e) {
        log.makeAlert(e, "Problem loading sink[%s] from disk.", schema.getDataSource())
           .addData("sinkDir", sinkDir)
           .emit();
      }
    }

    // Make sure we loaded all committed sinks.
    final Set<String> loadedSinks = Sets.newHashSet(
        Iterables.transform(sinks.keySet(), SegmentIdWithShardSpec::toString)
    );
    final Set<String> missingSinks = Sets.difference(committed.getHydrants().keySet(), loadedSinks);
    if (!missingSinks.isEmpty()) {
      throw new ISE("Missing committed sinks [%s]", Joiner.on(", ").join(missingSinks));
    }

    totalRows.set(rowsSoFar);
    return committed.getMetadata();
  }
221320
public static Class<? extends RowMapper> replaceRowMapping(Class returnTypeClass, Class<? extends RowMapper> rowMapperClass) {
        return _rowMappings.put(returnTypeClass, rowMapperClass);
    }
99595
public Sink update() {
    // [START update]
    Sink updatedSink = sink.toBuilder().setFilter("severity<=ERROR").build().update();
    // [END update]
    return updatedSink;
  }
741736
@Override
	public void accept(Visitor<Operator<?>> visitor) {
		for (GenericDataSinkBase<?> sink : this.sinks) {
			sink.accept(visitor);
		}
	}
564653
def rename(rename_map)
      @by_id.each.with_index do |klass, id|
        # Some entries can be nil. Ignore them.
        next unless klass

        if (new_name = rename_map[klass])
          # We have a rename request. Update the current @by_id entry.
          @by_id[id] = new_name
          # Remove the old class name from @by_class hash.
          @by_class.delete(klass)
          # Insert the new one with the current ID.
          @by_class[new_name] = id
        end
      end
    end
909885
@SuppressWarnings("unchecked")
  @Override
  public void sink(FlowProcess<? extends Configuration> flowProcess, SinkCall<Corc, OutputCollector> sinkCall)
      throws IOException {
    Corc corc = sinkCall.getContext();
    TupleEntry tupleEntry = sinkCall.getOutgoingEntry();
    for (Comparable<?> fieldName : tupleEntry.getFields()) {
      corc.set(fieldName.toString(), tupleEntry.getObject(fieldName));
    }
    sinkCall.getOutput().collect(null, corc);
  }
1276371
def build_sink(function: Callable[..., None] = None, *,
               unpack: bool = False):
    
    def _build_sink(function: Callable[..., None]):
        @wraps(function)
        def _wrapper(*args, **kwargs) -> Sink:
            if 'unpack' in kwargs:
                raise TypeError('"unpack" has to be defined by decorator')
            return Sink(function, *args, unpack=unpack, **kwargs)
        return _wrapper

    if function:
        return _build_sink(function)

    return _build_sink
584715
void shrink()
  {

    int n = m_opMap.elementAt(MAPINDEX_LENGTH);
    m_opMap.setToSize(n + 4);

    m_opMap.setElementAt(0,n);
    m_opMap.setElementAt(0,n+1);
    m_opMap.setElementAt(0,n+2);


    n = m_tokenQueue.size();
    m_tokenQueue.setToSize(n + 4);

    m_tokenQueue.setElementAt(null,n);
    m_tokenQueue.setElementAt(null,n + 1);
    m_tokenQueue.setElementAt(null,n + 2);
  }
1273896
public void addDataSink(GenericDataSinkBase<?> sink) {
		checkNotNull(sink, "The data sink must not be null.");
		
		if (!this.sinks.contains(sink)) {
			this.sinks.add(sink);
		}
	}
20397
def pipe(self, target):
        
        return self._replace(recver=self.recver.pipe(target))
1337740
def request_update(self, context):
        """"""
        pa_operation_unref(pa_context_get_sink_info_by_name(
            context, self.current_sink.encode(), self._sink_info_cb, None))
794893
private static void renameMethod(ClassNode buildee, MethodNode mNode, String newName) {
        buildee.addMethod(newName, mNode.getModifiers(), mNode.getReturnType(), mNode.getParameters(), mNode.getExceptions(), mNode.getCode());
        buildee.removeMethod(mNode);
    }
749735
private static LongConsumer adapt(Sink<Long> sink) {
        if (sink instanceof LongConsumer) {
            return (LongConsumer) sink;
        } else {
//            if (Tripwire.ENABLED)
//                Tripwire.trip(AbstractPipeline.class,
//                              "using LongStream.adapt(Sink<Long> s)");
            return sink::accept;
        }
    }
23216
private <T> QueryRunner<T> withPerSinkMetrics(
      final QueryRunner<T> sinkRunner,
      final QueryToolChest<T, ? extends Query<T>> queryToolChest,
      final SegmentId sinkSegmentId,
      final AtomicLong cpuTimeAccumulator
  )
  {
    // Note: reportSegmentAndCacheTime and reportSegmentTime are effectively the same here. They don't split apart
    // cache vs. non-cache due to the fact that Sinks may be partially cached and partially uncached. Making this
    // better would need to involve another accumulator like the cpuTimeAccumulator that we could share with the
    // sinkRunner.
    String sinkSegmentIdString = sinkSegmentId.toString();
    return CPUTimeMetricQueryRunner.safeBuild(
        new MetricsEmittingQueryRunner<>(
            emitter,
            queryToolChest,
            new MetricsEmittingQueryRunner<>(
                emitter,
                queryToolChest,
                sinkRunner,
                QueryMetrics::reportSegmentTime,
                queryMetrics -> queryMetrics.segment(sinkSegmentIdString)
            ),
            QueryMetrics::reportSegmentAndCacheTime,
            queryMetrics -> queryMetrics.segment(sinkSegmentIdString)
        ).withWaitMeasuredFromNow(),
        queryToolChest,
        emitter,
        cpuTimeAccumulator,
        false
    );
  }
1102136
def replace!(oth)
      if self.class != oth.class
        raise ArgumentError, "expected #{self.class} object"
      end

      component.each do |c|
        self.__send__("#{c}=", oth.__send__(c))
      end
    end
1248297
private <T> Collection<Integer> transformSink(SinkTransformation<T> sink) {

		Collection<Integer> inputIds = transform(sink.getInput());

		String slotSharingGroup = determineSlotSharingGroup(sink.getSlotSharingGroup(), inputIds);

		streamGraph.addSink(sink.getId(),
				slotSharingGroup,
				sink.getCoLocationGroupKey(),
				sink.getOperator(),
				sink.getInput().getOutputType(),
				null,
				"Sink: " + sink.getName());

		streamGraph.setParallelism(sink.getId(), sink.getParallelism());
		streamGraph.setMaxParallelism(sink.getId(), sink.getMaxParallelism());

		for (Integer inputId: inputIds) {
			streamGraph.addEdge(inputId,
					sink.getId(),
					0
			);
		}

		if (sink.getStateKeySelector() != null) {
			TypeSerializer<?> keySerializer = sink.getStateKeyType().createSerializer(env.getConfig());
			streamGraph.setOneInputStateKey(sink.getId(), sink.getStateKeySelector(), keySerializer);
		}

		return Collections.emptyList();
	}