@Override
    public Collection<TaskExecutorPartitionInfo> getTrackedPartitionsFor(JobID producingJobId) {
        return partitionTable.getTrackedPartitions(producingJobId).stream()
                .map(
                        partitionId -> {
                            final PartitionInfo<JobID, TaskExecutorPartitionInfo> partitionInfo =
                                    partitionInfos.get(partitionId);
                            Preconditions.checkNotNull(partitionInfo);
                            return partitionInfo.getMetaInfo();
                        })
                .collect(toList());
    }
@Test
    void testGetTrackedPartitionsFor() {
        final TestingShuffleEnvironment testingShuffleEnvironment = new TestingShuffleEnvironment();

        final JobID jobId = new JobID();
        final ResultPartitionID resultPartitionId = new ResultPartitionID();

        final TaskExecutorPartitionTracker partitionTracker =
                new TaskExecutorPartitionTrackerImpl(testingShuffleEnvironment);
        TaskExecutorPartitionInfo partitionInfo =
                new TaskExecutorPartitionInfo(
                        new TestingShuffleDescriptor(resultPartitionId),
                        new IntermediateDataSetID(),
                        1);

        partitionTracker.startTrackingPartition(jobId, partitionInfo);
        Collection<TaskExecutorPartitionInfo> partitions =
                partitionTracker.getTrackedPartitionsFor(jobId);

        assertThat(partitions).hasSize(1);
        assertThat(partitions.iterator().next()).isEqualTo(partitionInfo);
    }