Fold Consumer Groups into Sequin Stream sink

[acco]

• [X] Create new sink_consumer.sink for ConsumerGroupSink
• [X] Test end to end
• [X] Migrate http pull consumers into sink consumers