### Question:
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();  }### Answer:
@Test public void testConsumptionBasedPartitionRelease() { final ResultPartitionManager manager = new ResultPartitionManager(); final ResultPartition partition = new ResultPartitionBuilder() .setNumberOfSubpartitions(2) .isReleasedOnConsumption(true) .setResultPartitionManager(manager) .build(); manager.registerResultPartition(partition); partition.onConsumedSubpartition(0); assertFalse(partition.isReleased()); partition.onConsumedSubpartition(1); assertTrue(partition.isReleased()); }
@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()); }
@Test public void testReleaseAfterIdempotentCalls() { 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); partition.onConsumedSubpartition(1); assertTrue(partition.isReleased()); }