class ReleaseOnConsumptionResultPartition extends ResultPartition {

	@Override
	void onConsumedSubpartition(int subpartitionIndex) {
		if (isReleased()) {
			return;
		}

		final int remainingUnconsumed;

		// we synchronize only the bookkeeping section, to avoid holding the lock during any
		// calls into other components
		synchronized (lock) {
			if (consumedSubpartitions[subpartitionIndex]) {
				// repeated call - ignore
				return;
			}

			consumedSubpartitions[subpartitionIndex] = true;
			remainingUnconsumed = (--numUnconsumedSubpartitions);
		}

		LOG.debug("{}: Received consumed notification for subpartition {}.", this, subpartitionIndex);

		if (remainingUnconsumed == 0) {
			partitionManager.onConsumedPartition(this);
		} else if (remainingUnconsumed < 0) {
			throw new IllegalStateException("Received consume notification even though all subpartitions are already consumed.");
		}
	}

	  ReleaseOnConsumptionResultPartition(
			String owningTaskName,
			ResultPartitionID partitionId,
			ResultPartitionType partitionType,
			ResultSubpartition[] subpartitions,
			int numTargetKeyGroups,
			ResultPartitionManager partitionManager,
			FunctionWithException<BufferPoolOwner, BufferPool, IOException> bufferPoolFactory);

}