Refactored DefineProcessFunction into a few smaller pieces