implement shutdown for kafka consumer group