class CollectSinkFunction extends RichSinkFunction<IN> implements CheckpointedFunction, CheckpointListener {

	@Override
	public void invoke(IN value, Context context) throws Exception {
		bufferLock.lock();
		try {
			// TODO this implementation is not very effective,
			//  optimize this with MemorySegment if needed
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(baos);
			serializer.serialize(value, wrapper);
			invokingRecordBytes = baos.size();

			if (invokingRecordBytes > maxBytesPerBatch) {
				throw new RuntimeException(
					"Record size is too large for CollectSinkFunction. Record size is " + invokingRecordBytes + " bytes, " +
						"but max bytes per batch is only " + maxBytesPerBatch + " bytes. " +
						"Please consider increasing max bytes per batch value.");
			}

			if (currentBufferBytes + invokingRecordBytes > bufferSizeLimitBytes) {
				bufferCanAddNextResultCondition.await();
			}
			buffer.add(baos.toByteArray());
			currentBufferBytes += baos.size();
		} finally {
			bufferLock.unlock();
		}
	}

	public  CollectSinkFunction(TypeSerializer<IN> serializer, long maxBytesPerBatch, String accumulatorName);

}