StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.enableCheckpointing(true);

        Map<String, Object> sinkConfig = getSinkConfig();

        // Form msgIN & msgOUT Stream

        DataStream<DTO> stream1 = // consume from kafka topic 1
        KeyedStream<DTO, Tuple> stream1KeyedStream = stream1.keyBy("id");
        DataStream<DTO> stream2 = // consume from kafka topic 2
        KeyedStream<DTO, Tuple> stream2KeyedStream = stream2.keyBy("id");

        
        // Write Raw to S3
        //writeRawMessageToS3(stream1, stream2, sinkConfig);
        stream1.writeAsText("msgInRaw.txt", FileSystem.WriteMode.OVERWRITE);
        stream2.writeAsText("msgOutRaw.txt", FileSystem.WriteMode.OVERWRITE);

        // Union the two streams
        DataStream<DTO> unionStream = stream1KeyedStream.union(stream2KeyedStream);

        // Assign Timestamp and watermarks on the unionStream
        SingleOutputStreamOperator<DTO> timestampsAndWatermarksStream = unionStream.assignTimestampsAndWatermarks(new
                MessageTimestamp(Time.milliseconds(1500)));
        
        SingleOutputStreamOperator<Tuple2<DTO, DTO>> matchingAndNonMatchingMsgStream =
                timestampsAndWatermarksStream
                        .keyBy("some_key")
                        .timeWindow(Time.seconds(1))
                        .apply(new OuterJoinOperator());

        SingleOutputStreamOperator<Tuple2<DTO, DTO>> someDataStream =
                matchingAndNonMatchingMsgStream.flatMap(new SomeMessageExtractor());

        SplitStream<Tuple2<DTO, DTO>> split = someDataStream
                .split(new SomeErrorSplitter());

        DataStream<Tuple2<DTO, DTO>> errorStream = split.select("ErrorTuples");
        DataStream<Tuple2<DTO, DTO>> nonErrorStream = split
                .select("NonErrorTuples");

        
        // S3 in Error Data Folder
        //writeErrorMessageToS3(errorStream, sinkConfig);
        errorStream.writeAsText("errorStream.txt", FileSystem.WriteMode.OVERWRITE);
        // Non Fix Error Messages - Create generic Record --> Data Enrichment
        // --> S3 in Processed Data Folder
        SingleOutputStreamOperator<GenericRecord> genericRecordStream = nonErrorStream
                .flatMap(new GenericRecordExtractor());

        
        SplitStream<GenericRecord> valueErrorSplit = genericRecordStream.split(new ValueErrorSplitter());

        DataStream<GenericRecord> errorRecords = valueErrorSplit.select("ErrorRecords");
        DataStream<GenericRecord> nonErrorRecords = valueErrorSplit.select("NonErrorRecords");

        //errorRecords.addSink(createRollingSinkForErrorData(sinkConfig));
        errorRecords.writeAsText("errorRecords.txt", FileSystem.WriteMode.OVERWRITE);

        DataStream<GenericRecord> processedRecords = nonErrorRecords.map(new ReferenceDataExtractor());

        //writeOrderEntryMessageToS3(processedRecords, sinkConfig);
        processedRecords.writeAsText("processedRecords.txt", FileSystem.WriteMode.OVERWRITE);

        // Filter Orphan Message Stream
        DataStream<GenericRecord> nonMatchingRecords = processedRecords.filter(new NonMatchingFilter());
        
        //nonMatchingRecords.addSink(createRollingSinkForOrphanData(sinkConfig));
        nonMatchingRecords.writeAsText("nonMatchingRecords", FileSystem.WriteMode.OVERWRITE);

        env.execute();