public DataSourceTask(Environment environment) {
        super(environment);
    }
@Test
    void testDataSourceTask() throws IOException {
        int keyCnt = 100;
        int valCnt = 20;

        this.outList = new ArrayList<Record>();
        File tempTestFile = new File(tempFolder.toFile(), UUID.randomUUID().toString());
        InputFilePreparator.prepareInputFile(
                new UniformRecordGenerator(keyCnt, valCnt, false), tempTestFile, true);

        super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE);
        super.addOutput(this.outList);

        DataSourceTask<Record> testTask = new DataSourceTask<>(this.mockEnv);

        super.registerFileInputTask(
                testTask, MockInputFormat.class, tempTestFile.toURI().toString(), "\n");

        try {
            testTask.invoke();
        } catch (Exception e) {
            System.err.println(e);
            fail("Invoke method caused exception.");
        }

        try {
            Field formatField = DataSourceTask.class.getDeclaredField("format");
            formatField.setAccessible(true);
            MockInputFormat inputFormat = (MockInputFormat) formatField.get(testTask);
            assertThat(inputFormat.opened)
                    .withFailMessage(
                            "Invalid status of the input format. Expected for opened: true, Actual: %b",
                            inputFormat.opened)
                    .isTrue();
            assertThat(inputFormat.closed)
                    .withFailMessage(
                            "Invalid status of the input format. Expected for closed: true, Actual: %b",
                            inputFormat.closed)
                    .isTrue();
        } catch (Exception e) {
            System.err.println(e);
            fail("Reflection error while trying to validate inputFormat status.");
        }

        assertThat(this.outList)
                .withFailMessage(
                        "Invalid output size. Expected: %d, Actual: %d",
                        keyCnt * valCnt, outList.size())
                .hasSize(keyCnt * valCnt);

        HashMap<Integer, HashSet<Integer>> keyValueCountMap = new HashMap<>(keyCnt);

        for (Record kvp : this.outList) {

            int key = kvp.getField(0, IntValue.class).getValue();
            int val = kvp.getField(1, IntValue.class).getValue();

            if (!keyValueCountMap.containsKey(key)) {
                keyValueCountMap.put(key, new HashSet<Integer>());
            }
            keyValueCountMap.get(key).add(val);
        }

        assertThat(keyValueCountMap)
                .withFailMessage(
                        "Invalid key count in out file. Expected: %d, Actual: %d",
                        keyCnt, keyValueCountMap.size())
                .hasSize(keyCnt);

        for (Integer mapKey : keyValueCountMap.keySet()) {
            assertThat(keyValueCountMap.get(mapKey))
                    .withFailMessage(
                            "Invalid value count for key: %d. Expected: %d, Actual: %d",
                            mapKey, valCnt, keyValueCountMap.get(mapKey).size())
                    .hasSize(valCnt);
        }
    }