@Override
    public void startTrackingPartition(
            ResourceID producingTaskExecutorId,
            ResultPartitionDeploymentDescriptor resultPartitionDeploymentDescriptor) {
        Preconditions.checkNotNull(producingTaskExecutorId);
        Preconditions.checkNotNull(resultPartitionDeploymentDescriptor);

        // non-releaseByScheduler partitions don't require explicit partition release calls.
        if (!resultPartitionDeploymentDescriptor.getPartitionType().isReleaseByScheduler()) {
            return;
        }

        final ResultPartitionID resultPartitionId =
                resultPartitionDeploymentDescriptor.getShuffleDescriptor().getResultPartitionID();

        startTrackingPartition(
                producingTaskExecutorId, resultPartitionId, resultPartitionDeploymentDescriptor);
    }
@Test
    void testGetJobPartitionClusterPartition() {
        final TestingShuffleMaster shuffleMaster = new TestingShuffleMaster();

        final Queue<ReleaseCall> releaseCalls = new ArrayBlockingQueue<>(4);
        final Queue<PromoteCall> promoteCalls = new ArrayBlockingQueue<>(4);
        final JobMasterPartitionTrackerImpl partitionTracker =
                new JobMasterPartitionTrackerImpl(
                        new JobID(),
                        shuffleMaster,
                        resourceId ->
                                Optional.of(
                                        createTaskExecutorGateway(
                                                resourceId, releaseCalls, promoteCalls)));

        final ResourceID taskExecutorId = ResourceID.generate();
        final ResultPartitionID resultPartitionId1 = new ResultPartitionID();
        final ResultPartitionID resultPartitionId2 = new ResultPartitionID();

        final ResultPartitionDeploymentDescriptor clusterPartition =
                AbstractPartitionTrackerTest.createResultPartitionDeploymentDescriptor(
                        resultPartitionId1, ResultPartitionType.BLOCKING_PERSISTENT, false);
        final ResultPartitionDeploymentDescriptor jobPartition =
                AbstractPartitionTrackerTest.createResultPartitionDeploymentDescriptor(
                        resultPartitionId2, false);
        partitionTracker.startTrackingPartition(taskExecutorId, clusterPartition);
        partitionTracker.startTrackingPartition(taskExecutorId, jobPartition);

        assertThat(partitionTracker.getAllTrackedNonClusterPartitions())
                .containsExactly(jobPartition);
        assertThat(partitionTracker.getAllTrackedClusterPartitions())
                .containsExactly(clusterPartition);
    }