### 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)

If None is specified, requests will not be retried.
        :param timeout: (Optional) The amount of time, in seconds, to wait for the request
            to complete. Note that if retry is specified, the timeout applies to each
            individual attempt.
        :param metadata: (Optional) Additional metadata that is provided to the method.
        :return: A list of Pub/Sub ReceivedMessage objects each containing
            an ``ackId`` property and a ``message`` property, which includes
            the base64-encoded message content. See
            https://cloud.google.com/pubsub/docs/reference/rpc/google.pubsub.v1#google.pubsub.v1.ReceivedMessage
        """
        subscriber = await self._get_subscriber_client()
        subscription_path = f"projects/{project_id}/subscriptions/{subscription}"
        self.log.info("Pulling max %d messages from subscription (path) %s", max_messages, subscription_path)
        try:
            response = await subscriber.pull(
                request={
                    "subscription": subscription_path,
                    "max_messages": max_messages,
                    "return_immediately": return_immediately,
                },
                retry=retry,
                timeout=timeout,
                metadata=metadata,
            )
            result = getattr(response, "received_messages", [])
            self.log.info("Pulled %d messages from subscription (path) %s", len(result), subscription_path)
            return result
        except (HttpError, GoogleAPICallError) as e:
            raise PubSubException(f"Error pulling messages from subscription {subscription_path}", e)