### Basic Application Setup and Execution

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/application.md

This snippet demonstrates how to set up an Application instance, define a topic, create a StreamingDataFrame, apply a simple processing function, and then run the application. It's a common pattern for starting a Quix Streams application.

```python
from quixstreams import Application

# Set up an `app = Application` and `sdf = StreamingDataFrame`;
# add some operations to `sdf` and then run everything.

app = Application(broker_address='localhost:9092', consumer_group='group')
topic = app.topic('test-topic')
df = app.dataframe(topic)
df.apply(lambda value, context: print('New message', value))

app.run()
```

--------------------------------

### Local Neo4j Docker Setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/neo4j-sink.md

Instructions to run a local Neo4j instance using Docker for testing purposes. This includes the command to start the container and the connection details.

```bash
docker run --rm -d --name neo4j \
-p 7474:7474 \
-p 7687:7687 \
--env NEO4J_AUTH=neo4j/local_password \
neo4j:latest
```

--------------------------------

### IcebergSink Setup Example with AWS Glue

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Demonstrates how to configure and use the IcebergSink with an AWS-hosted Iceberg table and AWS Glue data catalog. Ensure AWS credentials are set up or provided.

```python
from quixstreams import Application
from quixstreams.sinks.community.iceberg import IcebergSink, AWSIcebergConfig

# Configure S3 bucket credentials
iceberg_config = AWSIcebergConfig(
    aws_s3_uri="", aws_region="", aws_access_key_id="", aws_secret_access_key=""
)

# Configure the sink to write data to S3 with the AWS Glue catalog spec
iceberg_sink = IcebergSink(
    table_name="glue.sink-test",
    config=iceberg_config,
    data_catalog_spec="aws_glue",
)

app = Application(broker_address='localhost:9092', auto_offset_reset="earliest")
topic = app.topic('sink_topic')

# Do some processing here
sdf = app.dataframe(topic=topic).print(metadata=True)

# Sink results to the IcebergSink
sdf.sink(iceberg_sink)


if __name__ == "__main__":
    # Start the application
    app.run()
```

--------------------------------

### Local MongoDB Connection Example

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/mongodb-sink.md

Example of configuring `MongoDBSink` to connect to a locally running MongoDB instance, typically started via Docker.

```python
MongoDBSink(
    host="localhost",
    # other required args here...
)
```

--------------------------------

### Install MkDocs and Themes

Source: https://github.com/quixio/quix-streams/blob/main/docs/build/README.md

Install MkDocs and its required themes for serving the documentation locally. Run this command in the `docs/build` directory.

```bash
python -m pip install mkdocs mkdocs-material mkdocs-material-extensions
```

--------------------------------

### Complete Producer Example

Source: https://github.com/quixio/quix-streams/blob/main/docs/producer.md

A full example demonstrating the creation of an Application, definition of a Topic with JSON serialization, and producing a message to Kafka.

```python
# Create an Application instance with Kafka configs
from quixstreams import Application


app = Application(
    broker_address='localhost:9092', consumer_group='example'
)

# Define a topic "my_topic" with JSON serialization
topic = app.topic(name='my_topic', value_serializer='json')

event = {"id": "1", "text": "Lorem ipsum dolor sit amet"}

# Create a Producer instance
with app.get_producer() as producer:
    
    # Serialize an event using the defined Topic 
    message = topic.serialize(key=event["id"], value=event)
    
    # Produce a message into the Kafka topic
    producer.produce(
        topic=topic.name, value=message.value, key=message.key
    )
```

--------------------------------

### Example Usage of KafkaReplicatorSource

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Demonstrates how to set up and run the KafkaReplicatorSource. This example configures an Application, initializes the source to connect to a 'second-kafka-topic' on 'localhost:9092', and then prints the resulting DataFrame.

```python
from quixstreams import Application
from quixstreams.sources.kafka import KafkaReplicatorSource

app = Application(
    consumer_group="group",
)

source = KafkaReplicatorSource(
    name="source-second-kafka",
    app_config=app.config,
    topic="second-kafka-topic",
    broker_address="localhost:9092",
)

sdf = app.dataframe(source=source)
sdf = sdf.print()
app.run()
```

--------------------------------

### Example Custom Source Implementation

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

An example demonstrating how to implement a custom source by extending the BaseSource class. This example shows a `RandomNumbersSource` that generates random numbers and produces them to a Kafka topic.

```APIDOC
## Example: RandomNumbersSource

### Description
This example shows a custom source implementation that generates random integers between 0 and 100 and publishes them to a Kafka topic. It demonstrates the implementation of `start`, `stop`, and `default_topic` methods required by `BaseSource`.

### Code
```python
import random
from quixstreams import Application, Topic
from quixstreams.sources.base import BaseSource

class RandomNumbersSource(BaseSource):
    def __init__(self):
        super().__init__()
        self._running = False

    def start(self):
        self._running = True
        while self._running:
            number = random.randint(0, 100)
            serialized = self._producer_topic.serialize(value=number)
            self._producer.produce(
                topic=self._producer_topic.name,
                key=serialized.key,
                value=serialized.value,
            )

    def stop(self):
        self._running = False

    def default_topic(self) -> Topic:
        return Topic(
            name="topic-name",
            value_deserializer="json",
            value_serializer="json",
        )

def main():
    app = Application(broker_address="localhost:9092")
    source = RandomNumbersSource()

    sdf = app.dataframe(source=source)
    sdf.print(metadata=True)

    app.run()

if __name__ == "__main__":
    main()
```
```

--------------------------------

### Install Build Requirements

Source: https://github.com/quixio/quix-streams/blob/main/docs/build/README.md

Install the necessary Python packages for building API documentation. Run this command in the `docs/build` directory.

```bash
python -m pip install -r requirements.txt
```

--------------------------------

### Install Quix Streams

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/local-file-source.md

Install the Quix Streams library using pip. No special options are required for basic installation.

```bash
pip install quixstreams
```

--------------------------------

### InfluxDB3Sink Initialization for Local Testing

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/influxdb3-sink.md

Example of how to initialize the InfluxDB3Sink connector to connect to a locally running InfluxDB3 instance. Ensure the host, organization_id, and token match your local setup.

```python
InfluxDB3Sink(
    host="http://localhost:8181",   # be sure to add http
    organization_id="local",        # unused, but required
    token="local",                  # unused, but required
)
```

--------------------------------

### Start Local Elasticsearch

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/elasticsearch-sink.md

Use this command to download and start a local Elasticsearch instance for testing purposes.

```bash
curl -fsSL https://elastic.co/start-local | sh
```

--------------------------------

### Source Start

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

Starts the source in a subprocess, marking it as running and executing its `run` method.

```APIDOC
## Source.start

### Description
This method is triggered in the subprocess when the source is started. It marks the source as running, execute it's run method and ensure cleanup happens.

### Method
start
```

--------------------------------

### Install Pub/Sub Source Dependency

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/google-cloud-pubsub-source.md

Install the necessary package to use the Pub/Sub source with Quix Streams.

```bash
pip install quixstreams[pubsub]
```

--------------------------------

### Install Pub/Sub Sink Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/google-cloud-pubsub-sink.md

Install the necessary dependencies for the Pub/Sub sink using pip.

```bash
pip install quixstreams[pubsub]
```

--------------------------------

### Install Development Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/CONTRIBUTING.md

Installs all necessary dependencies for development and testing, including those for the main project and testing utilities.

```bash
python3 -m pip install -r requirements.txt -r requirements-dev.txt -r tests/requirements.txt
```

--------------------------------

### FileSink Setup Method

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Abstract method to be implemented by subclasses for authenticating and validating the connection. This method is called during the setup phase of the sink.

```python
@abstractmethod
def setup()
    
```

--------------------------------

### Custom Reduce Aggregation Example

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Example demonstrating how to use the 'reduce()' method to calculate multiple aggregates (min, max, count) simultaneously within tumbling windows.

```python
sdf = StreamingDataFrame(...)

# Using "reduce()" to calculate multiple aggregates at once
def reducer(agg: dict, current: int):
    aggregated = {
        'min': min(agg['min'], current),
        'max': max(agg['max'], current),
        'count': agg['count'] + 1
    }
    return aggregated

def initializer(current) -> dict:
    return {'min': current, 'max': current, 'count': 1}

window = (
    sdf.tumbling_window(duration_ms=1000)
    .reduce(reducer=reducer, initializer=initializer)
    .final()
)
```

--------------------------------

### Install Quix Streams

Source: https://github.com/quixio/quix-streams/blob/main/README.md

Install Quix Streams using pip or conda. Requires Python 3.9+ and Apache Kafka 0.10+.

```shell
python -m pip install quixstreams
```

```shell
conda install -c conda-forge quixio::quixstreams
```

--------------------------------

### Start Local Kafka with Docker Compose

Source: https://github.com/quixio/quix-streams/blob/main/docs/tutorials/README.md

Start a local Kafka instance using Docker Compose. This command runs Kafka and its UI in the background.

```bash
docker-compose up -d
```

--------------------------------

### BaseSource.start

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

This method is triggered in the subprocess when the source is started. The subprocess will run as long as the start method executes. Use it to fetch data and produce it to Kafka.

```APIDOC
## BaseSource.start

### Description
Starts the source in a subprocess, fetching data and producing it to Kafka. The method should execute as long as the source needs to run.

### Method
`start`

### Returns
None
```

--------------------------------

### Get High Water Mark

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Retrieves the highest record event-time observed by any transaction on this partition since the process started. Returns None for a cold start.

```python
@property
def high_water_ms() -> Optional[int]:
    pass
```

--------------------------------

### BaseSource.setup Method Signature

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

The setup method is for initializing clients and performing any necessary validation before the source begins operation.

```python
def setup()
```

--------------------------------

### BaseSource.setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

When applicable, set up the client here along with any validation to affirm a valid/successful authentication/connection.

```APIDOC
## BaseSource.setup

### Description
Sets up the client and performs validation for authentication/connection when applicable.

### Method
`setup`

### Returns
None
```

--------------------------------

### Get Stream Root Path

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Retrieves a list of all parent Streams starting from the current Stream until the root nodes of the topology are reached.

```python
def root_path() -> List["Stream"]
```

--------------------------------

### Get Windows from WindowedPartitionTransaction

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Retrieves all windows that start within a specified time range and match a given prefix. Supports reverse ordering.

```python
def get_windows(start_from_ms: int,
                start_to_ms: int,
                prefix: bytes,
                backwards: bool = False) -> list[WindowDetail[V]]
```

--------------------------------

### QuixTSDataLakeSink.setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Initializes the blob storage client and tests the connection.

```APIDOC
## QuixTSDataLakeSink.setup

### Description
Initialize blob storage client and test connection.

### Method
POST

### Endpoint
N/A (Method Call)
```

--------------------------------

### IcebergSink Initialization and Usage

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Details on initializing and using the IcebergSink to write data to an Apache Iceberg table, including configuration options and an example setup.

```APIDOC
## IcebergSink

### Description
IcebergSink writes batches of data to an Apache Iceberg table. It serializes incoming data batches into Parquet format and appends them to the Iceberg table, updating the table schema as necessary. Currently supports AWS-hosted Iceberg with AWS Glue data catalog.

### Class Signature
```python
class IcebergSink(BatchingSink)
```

### `__init__` Method

#### Parameters

- **table_name** (str) - Required - The name of the Iceberg table.
- **config** (IcebergConfig) - Required - An IcebergConfig object with connection parameters.
- **data_catalog_spec** (str) - Required - The data catalog to use (e.g., "aws_glue" for AWS Glue).
- **schema** (Optional) - Optional - The Iceberg table schema. If None, a default schema is used.
- **partition_spec** (Optional) - Optional - The partition specification for the table. If None, a default is used.
- **on_client_connect_success** (Optional) - Optional - A callback made after successful client authentication, primarily for logging.
- **on_client_connect_failure** (Optional) - Optional - A callback made after failed client authentication. It accepts the raised Exception and must resolve or propagate it.

### Example Setup
```python
from quixstreams import Application
from quixstreams.sinks.community.iceberg import IcebergSink, AWSIcebergConfig

# Configure S3 bucket credentials
iceberg_config = AWSIcebergConfig(
    aws_s3_uri="", aws_region="", aws_access_key_id="", aws_secret_access_key=""
)

# Configure the sink to write data to S3 with the AWS Glue catalog spec
iceberg_sink = IcebergSink(
    table_name="glue.sink-test",
    config=iceberg_config,
    data_catalog_spec="aws_glue",
)

app = Application(broker_address='localhost:9092', auto_offset_reset="earliest")
topic = app.topic('sink_topic')

# Do some processing here
sdf = app.dataframe(topic=topic).print(metadata=True)

# Sink results to the IcebergSink
sdf.sink(iceberg_sink)

if __name__ == "__main__":
    # Start the application
    app.run()
```
```

--------------------------------

### Get Window Value

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Retrieves the value associated with a specific window defined by start and end timestamps. Returns a default value if the window is not found.

```python
def get_window(start_ms: int,
               end_ms: int,
               default: Any = None) -> Optional[Any]:
    pass
```

--------------------------------

### Run Tutorial Application

Source: https://github.com/quixio/quix-streams/blob/main/docs/tutorials/README.md

Execute the main application script for a tutorial. This script consumes and processes data from Kafka.

```bash
python ./path/to/tutorial_app.py
```

--------------------------------

### Composing StreamTimeoutTracker into a Custom Sink

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/stream-timeout-tracker.md

This example demonstrates how to integrate the StreamTimeoutTracker into a custom sink by wiring its lifecycle methods (setup, add, flush, stop) appropriately.

```APIDOC
## Custom Sink Integration Example

### Description
This snippet illustrates how to compose `StreamTimeoutTracker` within a custom sink, mirroring the lifecycle management pattern used in `QuixTSDataLakeSink`.

### Class: MyCustomSink
```python
from quixstreams.sinks.base import BatchingSink
from quixstreams.sinks.core.stream_timeout_tracker import StreamTimeoutTracker

class MyCustomSink(BatchingSink):
    def __init__(self, *, stream_timeout_ms=None, on_stream_timeout=None, **kwargs):
        super().__init__(**kwargs)
        self._timeout = StreamTimeoutTracker(
            stream_timeout_ms=stream_timeout_ms,
            on_stream_timeout=on_stream_timeout,
            thread_name="MyCustomSink-timeout-check",
        )

    def add(self, value, key, timestamp, headers, topic, partition, offset):
        super().add(value, key, timestamp, headers, topic, partition, offset)
        # Record activity for the stream
        self._timeout.touch(key, topic=topic, partition=partition, offset=offset)

    def flush(self):
        super().flush()
        # Perform a synchronous timeout check
        self._timeout.check_now()

    def setup(self):
        super().setup()
        # Start the background timeout checking thread
        self._timeout.start()

    def stop_timeout_tracker(self):
        # Explicitly stop the tracker thread
        self._timeout.stop()
```

**Lifecycle Management:**
- Call `setup()` to start the tracker's background thread.
- Call `add()` and pass the `key` to `_timeout.touch()` to mark stream activity.
- Call `flush()` and then `_timeout.check_now()` to perform checks.
- Call `stop()` (or `stop_timeout_tracker()` in this example) to gracefully shut down the tracker thread.

**Note:** If the sink's lifecycle is managed externally, ensure `tracker.stop()` is called during the host's shutdown process. Otherwise, the daemon thread may self-terminate after a period of inactivity.
```

--------------------------------

### Get Windowed State by Timestamp Range

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Retrieves the aggregated value for a window defined by start and end timestamps. Returns a default value if the window is not found.

```python
def get_window(start_ms: int,
               end_ms: int,
               default: Optional[V] = None) -> Optional[V]
```

--------------------------------

### BaseSink.setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Method to set up the client and perform validation for authentication/connection. Called during sink initialization.

```APIDOC
## BaseSink.setup

### Description
When applicable, set up the client here along with any validation to affirm a valid/successful authentication/connection.

### Method
setup

### Parameters
None
```

--------------------------------

### Composing StreamTimeoutTracker into a Custom Sink

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/stream-timeout-tracker.md

Integrate StreamTimeoutTracker into a custom sink by initializing it in `__init__`, calling `touch()` in `add()`, `check_now()` in `flush()`, `start()` in `setup()`, and `stop()` in a custom cleanup method.

```python
from quixstreams.sinks.base import BatchingSink
from quixstreams.sinks.core.stream_timeout_tracker import StreamTimeoutTracker


class MyCustomSink(BatchingSink):
    def __init__(self, *, stream_timeout_ms=None, on_stream_timeout=None, **kwargs):
        super().__init__(**kwargs)
        self._timeout = StreamTimeoutTracker(
            stream_timeout_ms=stream_timeout_ms,
            on_stream_timeout=on_stream_timeout,
            thread_name="MyCustomSink-timeout-check",
        )

    def add(self, value, key, timestamp, headers, topic, partition, offset):
        super().add(value, key, timestamp, headers, topic, partition, offset)
        self._timeout.touch(key, topic=topic, partition=partition, offset=offset)

    def flush(self):
        super().flush()
        self._timeout.check_now()

    def setup(self):
        super().setup()
        self._timeout.start()

    def stop_timeout_tracker(self):
        self._timeout.stop()
```

--------------------------------

### Serve MkDocs Documentation

Source: https://github.com/quixio/quix-streams/blob/main/docs/build/README.md

Start a local development server to view the MkDocs documentation. The configuration file `mkdocs.yml` is located one level up from the `docs/build` directory.

```bash
mkdocs serve -f ../../mkdocs.yml
```

--------------------------------

### Initialize and Use QuixEnvironmentSource

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Demonstrates how to initialize QuixEnvironmentSource to replicate topics from a Quix Cloud environment. Ensure you have the necessary Quix workspace ID and SDK token.

```python
from quixstreams import Application
from quixstreams.sources.kafka import QuixEnvironmentSource

app = Application(
    consumer_group="group",
)

source = QuixEnvironmentSource(
    name="source-quix",
    app_config=app.config,
    quix_workspace_id="WORKSPACE_ID",
    quix_sdk_token="WORKSPACE_SDK_TOKEN",
    topic="quix-source-topic",
)

sdf = app.dataframe(source=source)
sdf = sdf.print()
app.run()
```

--------------------------------

### Local InfluxDB3 Docker Setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/influxdb3-sink.md

Instructions for running a local InfluxDB3 instance using Docker for testing purposes. This command starts a container and exposes the InfluxDB3 service on port 8181.

```bash
docker run --rm -d --name influxdb3 \
-p 8181:8181 \
quay.io/influxdb/influxdb3-core:latest \
serve --node-id=host0 --object-store=memory
```

--------------------------------

### Initialize Application and DataFrame

Source: https://github.com/quixio/quix-streams/blob/main/docs/missing-data.md

Basic setup for using Quix Streams, including initializing the Application and creating a DataFrame. This is a prerequisite for using other DataFrame methods.

```python
from quixstreams import Application

# Initialize the Application
app = Application(...) 
sdf = app.dataframe(...)
```

--------------------------------

### KafkaReplicatorSink Usage Example

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

This snippet demonstrates how to initialize and use the KafkaReplicatorSink to send data to an external Kafka topic. Ensure you have the 'quixstreams' library installed and configure your Kafka broker address and topic name.

```python
from quixstreams import Application
from quixstreams.sinks.community.kafka import KafkaReplicatorSink

app = Application(
    consumer_group="group",
)

topic = app.topic("input-topic")

# Define the external Kafka cluster configuration
kafka_sink = KafkaReplicatorSink(
    broker_address="external-kafka:9092",
    topic_name="output-topic",
    value_serializer="json",
    key_serializer="bytes",
)

sdf = app.dataframe(topic=topic)
sdf.sink(kafka_sink)

app.run()
```

--------------------------------

### Initialize Pub/Sub Source

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

Set up a Google Cloud Pub/Sub source to read messages, forwarding them to Kafka with SDF transformations. Requires service account credentials and project/topic/subscription IDs. Note that message key forwarding is unsupported.

```python
from quixstreams import Application
from quixstreams.sources.community.pubsub import PubSubSource
from os import environ

source = PubSubSource(
    # Suggested: pass JSON-formatted credentials from an environment variable.
    service_account_json = environ["PUBSUB_SERVICE_ACCOUNT_JSON"],
    project_id="<project ID>",
    topic_id="<topic ID>",  # NOTE: NOT the full /x/y/z path!
    subscription_id="<subscription ID>",  # NOTE: NOT the full /x/y/z path!
    create_subscription=True,
)
app = Application(
    broker_address="localhost:9092",
    auto_offset_reset="earliest",
    consumer_group="gcp",
    loglevel="INFO"
)
sdf = app.dataframe(source=source).print(metadata=True)

if __name__ == "__main__":
    app.run()

```

--------------------------------

### Implement a Custom Random Numbers Source

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

Example of creating a custom source by extending BaseSource. This source generates random numbers and produces them to a Kafka topic. Ensure you implement start, stop, and default_topic methods.

```python
class RandomNumbersSource(BaseSource):
def __init__(self):
    super().__init__()
    self._running = False

def start(self):
    self._running = True

    while self._running:
        number = random.randint(0, 100)
        serialized = self._producer_topic.serialize(value=number)
        self._producer.produce(
            topic=self._producer_topic.name,
            key=serialized.key,
            value=serialized.value,
        )

def stop(self):
    self._running = False

def default_topic(self) -> Topic:
    return Topic(
        name="topic-name",
        value_deserializer="json",
        value_serializer="json",
    )


def main():
    app = Application(broker_address="localhost:9092")
    source = RandomNumbersSource()

    sdf = app.dataframe(source=source)
    sdf.print(metadata=True)

    app.run()


if __name__ == "__main__":
    main()
```

--------------------------------

### Create Quix Streams Application

Source: https://github.com/quixio/quix-streams/blob/main/docs/tutorials/solar-farm-enrichment/tutorial.md

Initializes a Quix Streams Application with broker address, consumer group, and offset reset configuration. Use 'earliest' to start reading from the beginning of the topic. Changelog topics are disabled for this example.

```python
import os
from quixstreams import Application

app = Application(
    broker_address=os.getenv("BROKER_ADDRESS", "localhost:9092"),
    consumer_group="temperature_alerter",
    auto_offset_reset="earliest",
    # Disable changelog topics for this app, but it's recommended to keep them "on" in production
    use_changelog_topics=False  
)
```

--------------------------------

### Configure and Use AzureFileSource

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/microsoft-azure-file-source.md

Instantiate and configure the AzureFileSource with specific parameters like filepath, container, connection string, and data transformation setters. This example demonstrates setting up a source for JSON files with gzip compression and replay speed.

```python
from quixstreams import Application
from quixstreams.sources.community.file.azure import AzureFileSource


def key_setter(record: dict) -> str:
    return record["host_id"]


def value_setter(record: dict) -> dict:
    return {k: record[k] for k in ["field_x", "field_y"]}


def timestamp_setter(record: dict) -> int:
    return record['timestamp']


source = AzureFileSource(
    filepath='folder_a/folder_b',
    container="<YOUR CONTAINER NAME>",
    connection_string="<YOUR CONNECTION STRING>",
    key_setter=key_setter,
    value_setter=value_setter,
    timestamp_setter=timestamp_setter,
    file_format="json",
    compression="gzip",
    has_partition_folders=False,
    replay_speed=0.5,
)
app = Application(
    broker_address="localhost:9092", 
    consumer_group='file-source',
    auto_offset_reset='latest',
)
app.add_source(source)


if __name__ == "__main__":
    app.run()
```

--------------------------------

### Custom Random Numbers Source Implementation

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

An example of a custom source inheriting from `BaseSource`. This source generates random numbers and produces them to a Kafka topic. It demonstrates the implementation of `start`, `stop`, and `default_topic` methods, and how to integrate it with an `Application` and `StreamingDataFrame`.

```python
class RandomNumbersSource(BaseSource):
def __init__(self):
    super().__init__()
    self._running = False

def start(self):
    self._running = True

    while self._running:
        number = random.randint(0, 100)
        serialized = self._producer_topic.serialize(value=number)
        self._producer.produce(
            topic=self._producer_topic.name,
            key=serialized.key,
            value=serialized.value,
        )

def stop(self):
    self._running = False

def default_topic(self) -> Topic:
    return Topic(
        name="topic-name",
        value_deserializer="json",
        value_serializer="json",
    )


def main():
    app = Application(broker_address="localhost:9092")
    source = RandomNumbersSource()

    sdf = app.dataframe(source=source)
    sdf.print(metadata=True)

    app.run()


if __name__ == "__main__":
    main()

```

--------------------------------

### Create StreamingDataFrame and Run Application

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/application.md

Set up an Application instance, define a topic, create a StreamingDataFrame from that topic, apply a transformation, and then run the application. This demonstrates a basic stream processing pipeline.

```python
from quixstreams import Application

# Set up an `app = Application` and  `sdf = StreamingDataFrame`;
# add some operations to `sdf` and then run everything.

app = Application(broker_address='localhost:9092', consumer_group='group')
topic = app.topic('test-topic')
df = app.dataframe(topic)
df.apply(lambda value, context: print('New message', value))

app.run()
```

--------------------------------

### RocksDBStore.__init__

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Initializes the RocksDBStore with specified configuration.

```APIDOC
## RocksDBStore.__init__

### Description
Initializes the RocksDBStore with specified configuration.

### Method Signature
```python
def __init__(
        name: str,
        stream_id: Optional[str],
        base_dir: str,
        changelog_producer_factory: Optional[ChangelogProducerFactory] = None,
        options: Optional[RocksDBOptionsType] = None)
```

### Arguments

- **name** (str) - A unique store name.
- **stream_id** (Optional[str]) - A topic name for this store.
- **base_dir** (str) - Path to a directory with the state.
- **changelog_producer_factory** (Optional[ChangelogProducerFactory]) - A ChangelogProducerFactory instance if using changelogs.
- **options** (Optional[RocksDBOptionsType]) - RocksDB options. If `None`, the default options will be used.
```

--------------------------------

### Install Quix DataLake Sink Dependency

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/quix-ts-datalake-sink.md

Install the necessary dependencies for the Quix DataLake Sink using pip. This command ensures that quixstreams is installed with the quixdatalake extra.

```commandline
pip install quixstreams[quixdatalake]
```

--------------------------------

### StateStoreManager.init

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Initializes the StateStoreManager and creates a store directory.

```APIDOC
## StateStoreManager.init

### Description
Initialize `StateStoreManager` and create a store directory.

### Method
```python
def init() -> None
```
```

--------------------------------

### Application Initialization with SDK Token

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/application.md

Use the `Application.Quix()` class method for initializing the application with a Quix SDK token. This is the recommended approach for Quix Cloud integration. Alternatively, set the `Quix__Sdk__Token` environment variable.

```python
app = Application.Quix(quix_sdk_token="YOUR_SDK_TOKEN")
```

--------------------------------

### Install Redis Sink Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/redis-sink.md

Install the necessary dependencies for the Redis sink connector using pip. This command ensures that all required packages, including those for Redis, are installed with the quixstreams package.

```command-line
pip install quixstreams[redis]
```

--------------------------------

### Initialize MQTT Source

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

Configure an MQTT source with connection details, authentication, and message handling options. Supports wildcard topics for flexible consumption.

```python
def __init__(
        topic: str,
        client_id: str,
        server: str,
        port: int,
        username: str = None,
        password: str = None,
        version: ProtocolVersion = "3.1.1",
        tls_enabled: bool = True,
        key_setter: MqttKeyValueSetter = _default_key_setter,
        value_setter: MqttKeyValueSetter = _default_value_setter,
        timestamp_setter: MqttTimestampSetter = _default_timestamp_setter,
        payload_deserializer: Optional[Callable[[Any],
                                                Any]] = _default_deserializer,
        qos: Literal[0, 1] = 1,
        on_client_connect_success: Optional[
            ClientConnectSuccessCallback] = None,
        on_client_connect_failure: Optional[
            ClientConnectFailureCallback] = None)

```

--------------------------------

### Protobuf Deserializer and Serializer Setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/advanced/schema-registry.md

Instantiate Protobuf deserializer and serializer with message types and Schema Registry configurations. Both can utilize client and serialization configurations.

```python
from quixstreams.models.serializers.protobuf import ProtobufDeserializer, ProtobufSerializer

from my_input_models_pb2 import InputProto
from my_output_models_pb2 import OutputProto

deserializer = ProtobufDeserializer(
    msg_type=InputProto,
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)
serializer = ProtobufSerializer(
    msg_type=OutputProto,
    schema_registry_client_config=schema_registry_client_config,
    schema_registry_serialization_config=schema_registry_serialization_config,
)
```

--------------------------------

### BaseSink Setup Method

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Method to set up the sink's client and perform validation, typically called during initialization.

```python
def setup()
```

--------------------------------

### Delete Windows by Max Start Time

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Removes windows from RocksDB that have a start time less than or equal to the specified maximum start time. Optionally deletes associated values if they can no longer belong to any active window.

```python
def delete_windows(max_start_time: int, delete_values: bool,
                   prefix: bytes) -> None:
```

--------------------------------

### Install Pandas Dependency

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/pandas-source.md

Install Quix Streams with the pandas optional dependency to use the PandasDataFrameSource.

```bash
pip install quixstreams[pandas]
```

--------------------------------

### Kafka Replicator Source Example

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/kafka-source.md

Demonstrates how to set up and use the KafkaReplicatorSource to read from a source topic and produce to a destination topic. Ensure the Application is configured with the desired processing guarantee.

```python
from quixstreams import Application
from quixstreams.sources.core.kafka import KafkaReplicatorSource

def main():
    app = Application()
    source = KafkaReplicatorSource(
      name="my-source",
      app_config=app.config,
      topic="source-topic",
      broker_address="source-broker-address"
    )

    sdf = app.dataframe(source=source)
    sdf.print(metadata=True)

    app.run()

if __name__ == "__main__":
    main()
```

--------------------------------

### Install Neo4j Sink Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/neo4j-sink.md

Install the necessary dependencies for the Neo4j sink using pip.

```bash
pip install quixstreams[neo4j]
```

--------------------------------

### Initialize and Use KinesisSource

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/amazon-kinesis-source.md

Demonstrates how to initialize KinesisSource with stream details and AWS credentials, and then integrate it with Quix Streams Application.

```python
from quixstreams import Application
from quixstreams.sources.community.kinesis import KinesisSource


kinesis = KinesisSource(
    stream_name="<YOUR STREAM>",
    aws_access_key_id="<YOUR KEY ID>",
    aws_secret_access_key="<YOUR SECRET KEY>",
    aws_region="<YOUR REGION>",
    auto_offset_reset="earliest",  # start from the beginning of the stream (vs end)
)

app = Application(
    broker_address="<YOUR BROKER INFO>",
    consumer_group="<YOUR GROUP>",
)

sdf = app.dataframe(source=kinesis).print(metadata=True)
# YOUR LOGIC HERE!

if __name__ == "__main__":
    app.run()
```

--------------------------------

### Instantiate and Use LocalFileSource

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/local-file-source.md

Example of how to import and instantiate a LocalFileSource with custom key, value, and timestamp setters, along with file format and compression settings. This source is then added to a Quix Streams Application.

```python
from quixstreams.sources.community.file.local import LocalFileSource


def key_setter(record: dict) -> str:
    return record["host_id"]


def value_setter(record: dict) -> dict:
    return {k: record[k] for k in ["field_x", "field_y"]}


def timestamp_setter(record: dict) -> int:
    return record['timestamp']


source = LocalFileSource(
    filepath='folder_a/folder_b',
    key_setter=key_setter,
    value_setter=value_setter,
    timestamp_setter=timestamp_setter,
    file_format="json",
    compression="gzip",
    has_partition_folders=False,
    replay_speed=0.5,
)
app = Application(
    broker_address="localhost:9092", 
    consumer_group='file-source',
    auto_offset_reset='latest',
)
app.add_source(source)


if __name__ == "__main__":
    app.run()
```

--------------------------------

### Custom StatefulSource Example

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Example of a custom `StatefulSource` implementation (`RandomNumbersSource`) that generates random numbers, manages state, produces messages, and flushes periodically. This example demonstrates how to integrate a custom source with the Quix Streams application.

```python
import random
import time

from quixstreams import Application
from quixstreams.sources import StatefulSource


class RandomNumbersSource(StatefulSource):
    def run(self):

        i = 0
        while self.running:
            previous = self.state.get("number", 0)
            current = random.randint(0, 100)
            self.state.set("number", current)

            serialized = self._producer_topic.serialize(value=current + previous)
            self.produce(key=str(current), value=serialized.value)
            time.sleep(0.5)

            # flush the state every 10 messages
            i += 1
            if i % 10 == 0:
                self.flush()


def main():
    app = Application(broker_address="localhost:9092")
    source = RandomNumbersSource(name="random-source")

    sdf = app.dataframe(source=source)
    sdf.print(metadata=True)

    app.run()


if __name__ == "__main__":
    main()
```

--------------------------------

### PubSubSink.__init__

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Initializes the PubSubSink with project and topic details, along with optional configuration for authentication, serialization, and callbacks.

```APIDOC
## PubSubSink.__init__

### Description
Initialize the PubSubSink.

### Method
__init__

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Arguments:
- `project_id` (str) - Required - GCP project ID.
- `topic_id` (str) - Required - Pub/Sub topic ID.
- `service_account_json` (Optional[str]) - Optional - an optional JSON string with service account credentials to connect to Pub/Sub. The internal `PublisherClient` will use the Application Default Credentials if not provided.
- `value_serializer` (Callable[[Any], Union[bytes, str]]) - Optional - Function to serialize the value to string or bytes (defaults to json.dumps).
- `key_serializer` (Callable[[Any], str]) - Optional - Function to serialize the key to string (defaults to bytes.decode).
- `flush_timeout` (int) - Optional - Defaults to 5.
- `on_client_connect_success` (Optional[ClientConnectSuccessCallback]) - Optional - An optional callback made after successful client authentication, primarily for additional logging.
- `on_client_connect_failure` (Optional[ClientConnectFailureCallback]) - Optional - An optional callback made after failed client authentication (which should raise an Exception). Callback should accept the raised Exception as an argument. Callback must resolve (or propagate/re-raise) the Exception.
- `**kwargs` - Additional keyword arguments passed to PublisherClient.
```

--------------------------------

### Install Kinesis Source Dependency

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/amazon-kinesis-source.md

Install the necessary package to use the Kinesis sink with Quix Streams.

```bash
pip install quixstreams[kinesis]
```

--------------------------------

### MQTTSink Initialization Parameters

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Shows the parameters required to initialize the MQTTSink, including connection details, authentication, serialization, and callbacks.

```python
def __init__(client_id: str,
             server: str,
             port: int,
             topic_root: str,
             username: str = None,
             password: str = None,
             version: ProtocolVersion = "3.1.1",
             tls_enabled: bool = True,
             key_serializer: Callable[[Any], str] = bytes.decode,
             value_serializer: Callable[[Any], str] = json.dumps,
             qos: Literal[0, 1] = 1,
             mqtt_flush_timeout_seconds: int = 10,
             retain: Union[bool, Callable[[Any], bool]] = False,
             properties: Optional[MqttPropertiesHandler] = None,
             on_client_connect_success: Optional[
                 ClientConnectSuccessCallback] = None,
             on_client_connect_failure: Optional[
                 ClientConnectFailureCallback] = None)

```

--------------------------------

### Basic MQTT Source Usage

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sources/mqtt-source.md

Demonstrates how to configure and use the MQTTSource with a Quix Streams application. Ensure your Kafka broker is accessible.

```python
from quixstreams import Application
from quixstreams.sources.community.mqtt import MQTTSource

mqtt_source = MQTTSource(
    topic="sensors/temperature",
    client_id="my-mqtt-client",
    server="mqtt.broker.com",
    port=1883,
    username="your_username",
    password="your_password",
    version="3.1.1",
)

app = Application(
    broker_address="localhost:9092",
    consumer_group="mqtt-consumer",
)

sdf = app.dataframe(source=mqtt_source).print(metadata=True)
# YOUR LOGIC HERE!

if __name__ == "__main__":
    app.run()
```

--------------------------------

### Install Elasticsearch Sink Dependency

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/elasticsearch-sink.md

Install the necessary package to use the Elasticsearch sink with Quix Streams.

```bash
pip install quixstreams[elasticsearch]
```

--------------------------------

### Example Usage

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Demonstrates how to integrate KinesisSource into a Quix Streams application.

```APIDOC
## Example Usage

```python
from quixstreams import Application
from quixstreams.sources.community.kinesis import KinesisSource


kinesis = KinesisSource(
    stream_name="<YOUR STREAM>",
    aws_access_key_id="<YOUR KEY ID>",
    aws_secret_access_key="<YOUR SECRET KEY>",
    aws_region="<YOUR REGION>",
    auto_offset_reset="earliest",  # start from the beginning of the stream (vs end)
)

app = Application(
    broker_address="<YOUR BROKER INFO>",
    consumer_group="<YOUR GROUP>",
)

sdf = app.dataframe(source=kinesis).print(metadata=True)
# YOUR LOGIC HERE!

if __name__ == "__main__":
    app.run()
```

**Note**: The incoming message value will be in bytes, so transform in your SDF accordingly.
```

--------------------------------

### Install MQTT Sink Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/mqtt-sink.md

Install the necessary dependencies for the MQTT Sink connector using pip.

```bash
pip install quixstreams[mqtt]
```

--------------------------------

### Install Amazon S3 Sink Dependency

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/amazon-s3-sink.md

Install the necessary package to use the S3 sink connector.

```bash
pip install quixstreams[s3]
```

--------------------------------

### Local PostgreSQL Docker Setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/postgresql-sink.md

Run a local PostgreSQL instance using Docker for testing purposes. Use the provided connection settings to configure the PostgreSQLSink.

```bash
docker run --rm -d --name postgres \
-e POSTGRES_PASSWORD=local \
-e POSTGRES_USER=local \
-e POSTGRES_DB=local \
-p 5432:5432 \
postgres
```

```python
PostgreSQLSink(
    host="localhost",
    port=5432,
    user="local",
    password="local",
    dbname="local",
    table_name="<YOUR TABLE NAME>",
)
```

--------------------------------

### Initialize Application with Environment Variables

Source: https://github.com/quixio/quix-streams/blob/main/docs/quix-platform.md

Recommended approach for external connections: set `Quix__Sdk__Token` and `Quix__Portal__Api` environment variables. These are automatically configured in Quix Cloud and by the Quix CLI.

```python
from quixstreams import Application
import os

app = Application(broker_address=os.getenv("YOUR_ENV_VAR", None))
```

--------------------------------

### Install MongoDB Sink Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/mongodb-sink.md

Install the necessary dependencies for the MongoDB sink connector using pip.

```bash
pip install quixstreams[mongodb]
```

--------------------------------

### Install Commit Pre-hook

Source: https://github.com/quixio/quix-streams/blob/main/CONTRIBUTING.md

Installs the pre-commit hook to enforce code style and quality checks before committing.

```bash
pre-commit install
```

--------------------------------

### Source Initialization

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

Initializes a new Source instance with essential configuration parameters.

```APIDOC
## Source.__init__

### Description
Initializes a new Source instance with essential configuration parameters.

### Method
__init__

### Parameters
#### Path Parameters
- **name** (str) - Required - The source unique name. It is used to generate the topic configuration.
- **shutdown_timeout** (float) - Optional - Time in second the application waits for the source to gracefully shutdown. Defaults to 10.
- **on_client_connect_success** (Optional[ClientConnectSuccessCallback]) - Optional - An optional callback made after successful client authentication, primarily for additional logging.
- **on_client_connect_failure** (Optional[ClientConnectFailureCallback]) - Optional - An optional callback made after failed client authentication (which should raise an Exception). Callback should accept the raised Exception as an argument. Callback must resolve (or propagate/re-raise) the Exception.
```

--------------------------------

### Example Record Processed by SDF

Source: https://github.com/quixio/quix-streams/blob/main/docs/tutorials/websocket-source/tutorial.md

An example of a raw record processed by the StreamingDataFrame, showing its structure before transformation.

```python
{ 'value': { 'type': 'ticker',
             'sequence': 754296790,
             'product_id': 'ETH-BTC',
             'price': '0.00005',
             'open_24h': '0.00008',
             'volume_24h': '322206074.45925051',
             'low_24h': '0.00005',
             'high_24h': '0.00041',
             'volume_30d': '3131937035.46099349',
             'best_bid': '0.00001',
             'best_bid_size': '1000000000.00000000',
             'best_ask': '0.00006',
             'best_ask_size': '166668.66666667',
             'side': 'sell',
             'time': '2024-09-19T10:01:26.411029Z',
             'trade_id': 28157206,
             'last_size': '16666.86666667'}}
```

--------------------------------

### Initialize QuixTSDataLakeCatalogClient

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Initializes the catalog client with a base URL and an optional authentication token. Use this to set up your connection to the Catalog API.

```python
class QuixTSDataLakeCatalogClient()

```

```python
def __init__(base_url: str, auth_token: Optional[str] = None)
```

--------------------------------

### Install Azure File Sink Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/microsoft-azure-file-sink.md

Install the necessary dependencies for the Azure File Sink using pip.

```bash
pip install quixstreams[azure-file]
```

--------------------------------

### FileSink.setup

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sinks.md

Abstract method to authenticate and validate the connection for the FileSink.

```APIDOC
## FileSink.setup

### Description
Authenticates and validates the connection for the FileSink.

### Method
setup

### Parameters
This method does not accept any arguments.
```

--------------------------------

### Install Iceberg Sink Dependencies

Source: https://github.com/quixio/quix-streams/blob/main/docs/connectors/sinks/apache-iceberg-sink.md

Install the necessary dependencies for the IcebergSink when using the AWS Glue data catalog.

```commandline
# To use IcebergSink with AWS Glue data catalog
pip install quixstreams[iceberg_aws]
```

--------------------------------

### Run Tutorial Producer

Source: https://github.com/quixio/quix-streams/blob/main/docs/tutorials/README.md

Execute the producer script for a tutorial. This typically sends data to Kafka.

```bash
python ./path/to/producer.py
```

--------------------------------

### Install Project in Editable Mode

Source: https://github.com/quixio/quix-streams/blob/main/CONTRIBUTING.md

Installs the project in editable mode, allowing local changes to be used as a module without re-installation.

```bash
python3 -m pip install --editable .
```

--------------------------------

### Initialize PubSub Source

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/sources.md

Configure a Pub/Sub source for ingesting data. Specify project, topic, and subscription IDs. Optional parameters control commit behavior, message ordering, and client connection callbacks.

```python
def __init__(project_id: str,
             topic_id: str,
             subscription_id: str,
             service_account_json: Optional[str] = None,
             commit_every: int = 100,
             commit_interval: float = 5.0,
             create_subscription: bool = False,
             enable_message_ordering: bool = False,
             shutdown_timeout: float = 10.0,
             on_client_connect_success: Optional[
                 ClientConnectSuccessCallback] = None,
             on_client_connect_failure: Optional[
                 ClientConnectFailureCallback] = None)
```

--------------------------------

### RocksDBStorePartition.high_water_ms

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Retrieves the highest record event-time observed by any transaction on this partition since the process started. Returns None for a cold start.

```APIDOC
## Property RocksDBStorePartition.high_water_ms

### Description
Highest record event-time observed by any transaction on this partition since the process started, or ``None`` for cold start.

### Returns
- `Optional[int]`: The highest observed event time in milliseconds, or None.
```

--------------------------------

### Source.__init__

Source: https://github.com/quixio/quix-streams/blob/main/docs/api-reference/quixstreams.md

Initializes a new Source instance. This is the constructor for creating custom sources.

```APIDOC
## Source.__init__

### Description
Initializes a new Source instance. This is the constructor for creating custom sources.

### Method
__init__

### Parameters
#### Arguments
- **name** (str) - The source unique name. It is used to generate the topic configuration.
- **shutdown_timeout** (float) - Optional - Time in second the application waits for the source to gracefully shutdown. Defaults to 10.
- **on_client_connect_success** (Optional[ClientConnectSuccessCallback]) - Optional - An optional callback made after successful client authentication, primarily for additional logging.
- **on_client_connect_failure** (Optional[ClientConnectFailureCallback]) - Optional - An optional callback made after failed client authentication (which should raise an Exception). Callback should accept the raised Exception as an argument. Callback must resolve (or propagate/re-raise) the Exception.
```