public DataSinkTask(Environment environment) {
        super(environment);
    }
@Test
    void testDataSinkTask() {
        FileReader fr = null;
        BufferedReader br = null;
        try {
            int keyCnt = 100;
            int valCnt = 20;

            super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE);
            super.addInput(new UniformRecordGenerator(keyCnt, valCnt, false), 0);

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

            File tempTestFile = new File(tempFolder.toFile(), UUID.randomUUID().toString());
            super.registerFileOutputTask(
                    MockOutputFormat.class, tempTestFile.toURI().toString(), new Configuration());

            testTask.invoke();

            assertThat(tempTestFile).withFailMessage("Temp output file does not exist").exists();

            fr = new FileReader(tempTestFile);
            br = new BufferedReader(fr);

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

            while (br.ready()) {
                String line = br.readLine();

                Integer key = Integer.parseInt(line.substring(0, line.indexOf("_")));
                Integer val =
                        Integer.parseInt(line.substring(line.indexOf("_") + 1, line.length()));

                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 key : keyValueCountMap.keySet()) {
                assertThat(keyValueCountMap.get(key))
                        .withFailMessage(
                                "Invalid value count for key: %d. Expected: %d Actual: %d",
                                key, valCnt, keyValueCountMap.get(key).size())
                        .hasSize(valCnt);
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (Throwable t) {
                }
            }
            if (fr != null) {
                try {
                    fr.close();
                } catch (Throwable t) {
                }
            }
        }
    }