@Override
    public void start(Map<String, String> props) {
        AbstractConfig config = new AbstractConfig(FileStreamSinkConnector.CONFIG_DEF, props);
        filename = config.getString(FileStreamSinkConnector.FILE_CONFIG);
        if (filename == null || filename.isEmpty()) {
            outputStream = System.out;
        } else {
            try {
                outputStream = new PrintStream(
                    Files.newOutputStream(Paths.get(filename), StandardOpenOption.CREATE, StandardOpenOption.APPEND),
                    false,
                    StandardCharsets.UTF_8.name());
            } catch (IOException e) {
                throw new ConnectException("Couldn't find or create file '" + filename + "' for FileStreamSinkTask", e);
            }
        }
    }
@Test
    public void testStart() throws IOException {
        task = new FileStreamSinkTask();
        Map<String, String> props = new HashMap<>();
        props.put(FileStreamSinkConnector.FILE_CONFIG, outputFile);
        task.start(props);

        HashMap<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        task.put(Collections.singletonList(
                new SinkRecord("topic1", 0, null, null, Schema.STRING_SCHEMA, "line0", 1)
        ));
        offsets.put(new TopicPartition("topic1", 0), new OffsetAndMetadata(1L));
        task.flush(offsets);

        int numLines = 3;
        String[] lines = new String[numLines];
        int i = 0;
        try (BufferedReader reader = Files.newBufferedReader(Paths.get(outputFile))) {
            lines[i++] = reader.readLine();
            task.put(Arrays.asList(
                    new SinkRecord("topic1", 0, null, null, Schema.STRING_SCHEMA, "line1", 2),
                    new SinkRecord("topic2", 0, null, null, Schema.STRING_SCHEMA, "line2", 1)
            ));
            offsets.put(new TopicPartition("topic1", 0), new OffsetAndMetadata(2L));
            offsets.put(new TopicPartition("topic2", 0), new OffsetAndMetadata(1L));
            task.flush(offsets);
            lines[i++] = reader.readLine();
            lines[i++] = reader.readLine();
        }

        while (--i >= 0) {
            assertEquals("line" + i, lines[i]);
        }
    }