Closes the stream with a sink operator.

def sink(self):
        """Closes the stream with a sink operator."""
        op = Operator(
            _generate_uuid(),
            OpType.Sink,
            "Sink",
            num_instances=self.env.config.parallelism)
        return self.__register(op)