@Test public void testMultipleReleaseCallsAreIdempotent() { final ResultPartitionManager manager = new ResultPartitionManager(); final ResultPartition partition = new ResultPartitionBuilder() .setNumberOfSubpartitions(2) .isReleasedOnConsumption(true) .setResultPartitionManager(manager) .build(); manager.registerResultPartition(partition); partition.onConsumedSubpartition(0); partition.onConsumedSubpartition(0); assertFalse(partition.isReleased()); }
@Override void onConsumedSubpartition(int subpartitionIndex) { if (isReleased()) { return; } final int remainingUnconsumed; synchronized (lock) { if (consumedSubpartitions[subpartitionIndex]) { 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 extends ResultPartition { @Override void onConsumedSubpartition(int subpartitionIndex) { if (isReleased()) { return; } final int remainingUnconsumed; synchronized (lock) { if (consumedSubpartitions[subpartitionIndex]) { 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 extends ResultPartition { @Override void onConsumedSubpartition(int subpartitionIndex) { if (isReleased()) { return; } final int remainingUnconsumed; synchronized (lock) { if (consumedSubpartitions[subpartitionIndex]) { 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); }
ReleaseOnConsumptionResultPartition extends ResultPartition { @Override void onConsumedSubpartition(int subpartitionIndex) { if (isReleased()) { return; } final int remainingUnconsumed; synchronized (lock) { if (consumedSubpartitions[subpartitionIndex]) { 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); @Override ResultSubpartitionView createSubpartitionView(int index, BufferAvailabilityListener availabilityListener); @Override String toString(); }
ReleaseOnConsumptionResultPartition extends ResultPartition { @Override void onConsumedSubpartition(int subpartitionIndex) { if (isReleased()) { return; } final int remainingUnconsumed; synchronized (lock) { if (consumedSubpartitions[subpartitionIndex]) { 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); @Override ResultSubpartitionView createSubpartitionView(int index, BufferAvailabilityListener availabilityListener); @Override String toString();  }