text: Title: Question: How to provide oauth_cb to confluent kafka consumer/producer
  Body: We are trying to set up a faststream application which connects via IAM to an MSK cluster (confluent kafka). We got a successful connection to work by directly creating a `Consumer` and now want to use the same settings in faststream.
  
  **Working code**
  
  ```python
  from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
  from confluent_kafka import Consumer
  
  
  def oauth_cb(oauth_config):
      auth_token, expiry_ms = MSKAuthTokenProvider.generate_auth_token_from_role_arn("us-east-1", "<assumed_role_arn>")
      return auth_token, expiry_ms / 1000
  
  
  config = {
      "bootstrap.servers": "bootstrap.servers",
      "security.protocol": "SASL_SSL",
      "sasl.mechanism": "OAUTHBEARER",
      "oauth_cb": oauth_cb,
      "group.id": "our-group",
  }
  
  c = Consumer(config)
  c.poll(0)
  print(c.list_topics(timeout=1).topics)
  ```
  
  A callback is added to the producer via `oauth_cb` setting which generates a new token using [aws-msk-iam-sasl-signer-python](https://github.com/aws/aws-msk-iam-sasl-signer-python). The setting can't be found in the configuration file from [librdkafka](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md) but in the documentation for [confluent_kafka](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#kafka-client-configuration).
  
  > oauth_cb(config_str): Callback for retrieving OAuth Bearer token. Function argument config_str is a str from config: sasl.oauthbearer.config. Return value of this callback is expected to be (token_str, expiry_time) tuple where expiry_time is the time in seconds since the epoch as a floating point number. This callback is useful only when sasl.mechanisms=OAUTHBEARER is set and is served to get the initial token before a successful broker connection can be made. The callback can be triggered by calling client.poll() or producer.flush().
  
  Calling `.poll()` is required to invoke the callback for the first time (see [this bug](https://github.com/confluentinc/confluent-kafka-python/issues/1713)).  Without the token, a KafkaException is raised:
  
  ```
  cimpl.KafkaException: KafkaError{code=_TRANSPORT,val=-195,str="Failed to get metadata: Local: Broker transport failure"}
  ```
  
  Is there a way to pass the `oauth_cb` to the kafka consumer/producer in faststream? Or should we use a different configuration like `oauthbearer_token_refresh_cb`?
  
  ```python
  broker = KafkaBroker(
      "localhost:9098",
      security=security,
      config=ConfluentConfig(
          {
              "sasl.mechanism": "OAUTHBEARER",
              "oauthbearer_token_refresh_cb": oauth_cb,
          }
      ),
  )
  ```
  
  Above code fails with a KafkaError:
  
  ```
  cimpl.KafkaException: KafkaError{code=_INVALID_ARG,val=-186,str="Property "oauthbearer_token_refresh_cb" must be set through dedicated .._set_..() function"}
  ```
  
  Any help would be very welcome :)