### Description
Re: Google Cloud Provider -- PubSub

The PubSub [Topics Create API](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/create#request-body) supports the SchemaSettings and messageRetentionDuration parameters. These should also be supported via the [PubSubCreateTopicOperator](https://github.com/apache/airflow/blob/9207e7d5e5c183d2e63c3030216b14709257668e/airflow/providers/google/cloud/operators/pubsub.py#L50). Thanks!

### Use case/motivation

_No response_

### Related issues

_No response_

### Are you willing to submit a PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

self,
        topic: str,
        messages: list[dict],
        project_id: str = PROVIDE_PROJECT_ID,
    ) -> None:
        """
        Publishes messages to a Pub/Sub topic.
        :param topic: the Pub/Sub topic to which to publish; do not
            include the ``projects/{project}/topics/`` prefix.
        :param messages: messages to publish; if the data field in a
            message is set, it should be a bytestring (utf-8 encoded)
            https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#pubsubmessage
        :param project_id: Optional, the Google Cloud project ID in which to publish.
            If set to None or missing, the default project_id from the Google Cloud connection is used.
        """
        self._validate_messages(messages)
        publisher = self.get_conn()
        topic_path = f"projects/{project_id}/topics/{topic}"
        self.log.info("Publish %d messages to topic (path) %s", len(messages), topic_path)
        try:
            for message in messages:
                future = publisher.publish(
                    topic=topic_path, data=message.get("data", b""), **message.get("attributes", {})
                )
                future.result()
        except GoogleAPICallError as e:
            raise PubSubException(f"Error publishing to topic {topic_path}", e)
        self.log.info("Published %d messages to topic (path) %s", len(messages), topic_path)
    @staticmethod
    def _validate_messages(messages) -> None: