<a id="connect-firebase-source-connector"></a>

# Google Firebase Source Connector for Confluent Platform

The Kafka Connect Google Firebase Source connector enables users to read data
from a [Google Firebase Realtime Database](https://firebase.google.com/docs/database)  and persist the data in Apache Kafka®
topics.  For more information, refer to [Firebase Source connector
configuration properties](firebase_source_connector_config.md#firebase-source-connector-config).

#### IMPORTANT
<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
[Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
are no longer available for download.

## Features

### At least once delivery

This connector guarantees that records are delivered at least once to the Kafka
topic. If the connector restarts, there may be some duplicate records in the
Kafka topic.

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

See [Confluent Platform license](firebase_source_connector_config.md#firebase-source-license-config) for license properties and
[Confluent License Properties](firebase_source_connector_config.md#firebase-source-license-topic-configuration) for information about the license
topic.

## Limitations

The Firebase Source connector has the following limitation:

- **Unable to trace changes in data-reference when offline**: When the
  connector is not running, the records which are added or updated (multiple
  times) from the data reference will not get captured after connector restarts.
  A complete snapshot is taken each time the connector restarts, when the
  configuration `gcp.firebase.snapshot` is set to `true`. This may lead to
  the connector writing duplicate records to the Kafka topic.

## Configuration Properties

For a complete list of configuration properties for the source connector, see
[Configuration Reference for Firebase Source Connector for Confluent Platform](firebase_source_connector_config.md#firebase-source-connector-config).

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

## Install the Google Firebase Source Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- If you want to install the connector using Confluent Marketplace, you must install
  the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This
  is installed by default with Confluent Enterprise.
- Kafka Broker: Confluent Platform 3.3.0 or above, or Kafka 0.11.0 or above
- Connect: Confluent Platform 4.0.0 or above, or Kafka 1.0.0 or above
- Java 1.8
- A GCP service account with role `Firebase Realtime Database Viewer` for
  source connector and `Firebase Realtime Database Admin` for sink connector is
  required. You can create this service account in the [Google Cloud Console](https://console.cloud.google.com/iam-admin). For creating a service account,
  refer [creating and managing service accounts](https://cloud.google.com/iam/docs/creating-managing-service-accounts).

### Install the connector using the Confluent CLI

To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory
and run the following command:

```bash
confluent connect plugin install confluentinc/kafka-connect-firebase:latest
```

You can install a specific version by replacing `latest` with a version number
as shown in the following example:

```bash
confluent connect plugin install confluentinc/kafka-connect-firebase:1.2.0
```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-firebase) for your
connector and then follow the manual connector installation [instructions](https://docs.confluent.io/home/connect/install.html).

## Quick Start

In this Quick Start, you configure the Firebase Source connector to capture
records from a Firebase database  and write them to a Kafka topic.

### Start Confluent

Start the Confluent services using the following [Confluent
CLI](https://docs.confluent.io/confluent-cli/current/index.html) command:

```bash
confluent local start
```

#### IMPORTANT
Do not use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production
environments.

### Property-based example

Create a configuration file `firebase-source.properties` with the following
content. This file should be placed inside the Confluent Platform installation directory. This
configuration is used typically along with [standalone
workers](/platform/current/connect/concepts.html#standalone-workers).

```text
name=FirebaseSourceConnector

connector.class=io.confluent.connect.firebase.FirebaseSourceConnector
tasks.max=1

gcp.firebase.credentials.path=file-path-to-your-gcp-service-account-json-file
gcp.firebase.database.reference=https://<gcp-project-id>.firebaseio.com
gcp.firebase.snapshot=true

confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1
confluent.license=
```

Run the connector with this configuration.

```bash
confluent local load FirebaseSourceConnector --config firebase-source.properties
```

The output should resemble:

```json
{
    "name":"FirebaseSourceConnector",
    "config":{
        "tasks.max":"1",
        "connector.class":"io.confluent.connect.firebase.FirebaseSourceConnector",
        "gcp.firebase.credentials.path":"file-path-to-your-gcp-service-account-json-file",
        "gcp.firebase.database.reference":"https://<gcp-project-id>.firebaseio.com",
        "gcp.firebase.snapshot":"true",
        "confluent.topic.bootstrap.servers":"localhost:9092",
        "confluent.topic.replication.factor":"1",
         "name":"FirebaseSourceConnector"
     },
    "tasks":[
        {
            "connector":"FirebaseSourceConnector",
            "task":0
        }
    ],
    "type":"source"
}
```

Confirm that the connector is in a `RUNNING` state by running the following
command:

```bash
confluent local status FirebaseSourceConnector
```

The output should resemble:

```bash
{
   "name":"FirebaseSourceConnector",
   "connector":{
      "state":"RUNNING",
      "worker_id":"127.0.1.1:8083"
   },
   "tasks":[
      {
         "id":0,
         "state":"RUNNING",
         "worker_id":"127.0.1.1:8083"
      }
   ],
   "type":"source"
}
```

### REST-based example

Use this setting with [distributed
workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON to
`config.json`, configure all of the required values, and use the following
command to post the configuration to one of the distributed connect workers. For
more information about the Kafka Connect REST API, see [this
documentation](/platform/current/connect/references/restapi.html).

```json
{
  "name" : "FirebaseSourceConnector",
  "config" : {
    "connector.class" : "io.confluent.connect.firebase.FirebaseSourceConnector",
    "tasks.max" : "1",

    "gcp.firebase.credentials.path" : "file-path-to-your-gcp-service-account-json-file",
    "gcp.firebase.database.reference": "https://<gcp-project-id>.firebaseio.com",
    "gcp.firebase.snapshot" : "true",

    "confluent.topic.bootstrap.servers": "localhost:9092",
    "confluent.topic.replication.factor": "1",
    "confluent.license": " Omit to enable trial mode "
  }
}
```

#### NOTE
Change the `confluent.topic.bootstrap.servers` property to include your
broker address(es), and change the `confluent.topic.replication.factor` to
`3` for staging or production use.

Use curl to post a configuration to one of the Connect workers. Change
`http://localhost:8083/` to the endpoint of one of your Connect worker(s).

```bash
curl -sS -X POST -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors
```

Use the following command to update the configuration of existing connector.

```bash
curl -s -X PUT -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors/FirebaseSourceConnector/config
```

Confirm that the connector is in a `RUNNING` state by running the following
command:

```bash
curl http://localhost:8083/connectors/FirebaseSourceConnector/status
```

The output should resemble:

```bash
{
   "name":"FirebaseSourceConnector",
   "connector":{
      "state":"RUNNING",
      "worker_id":"127.0.1.1:8083"
   },
   "tasks":[
      {
         "id":0,
         "state":"RUNNING",
         "worker_id":"127.0.1.1:8083"
      }
   ],
   "type":"source"
}
```

To publish records into Firebase, follow the [Firebase documentation](https://firebase.google.com/docs/database/admin/save-data). The data
produced to firebase should adhere to the following [data format](#firebase-source-data-format). You can also use the JSON example mentioned in
the [data format section](#firebase-source-data-format), save it into a
`data.json` file and finally import it into a Firebase database reference using the
import feature in the Firebase console.

To consume records written by the connector to the Kafka topic, run the following
command:

```bash
kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081  --topic artists --from-beginning
```

```bash
kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081  --topic songs --from-beginning
```

<a id="firebase-source-data-format"></a>

## Firebase Database Structure

The connector expects the Firebase database to have data in the following format:

```bash
{
   "musicBlog":{
      "artists":{
         "artistId1":{
            "name":"Michael Jackson",
            "genre":"Pop"
         },
         "artistId2":{
            "name":"Bob Dylan",
            "genre":"American folk"
         },
         "artistId3":{
            "name":"Freddie Mercury",
            "genre":"Rock"
         }
      },
      "songs":{
         "songId1":{
            "title":"billie jean",
            "artist":"Michael Jackson"
         },
         "songId2":{
            "title":"hurricane",
            "artist":"Bob Dylan"
         },
         "songId3":{
            "title":"bohemian rhapsody",
            "artist":"Freddie Mercury"
         }
      }
   }
}
```

If the `gcp.firebase.database.reference` is configured to
`{firebase-application-url}/musicBlog`, the connector will realize `artists`
and `songs` as kafka topic and the records within those topics will be
processed and pushed to respective kafka topic named `artists` and `songs`.

#### NOTE
The `key/recordId` for records within the Firebase database collection
should follow the [naming conventions](https://avro.apache.org/docs/1.8.1/spec.html#names) for Avro fields.

## Record Schema

The source connector creates Kafka records in the following format:

* **Kafka Key**: The Kafka record key consists of firebase record ID and is of
  type string schema.
* **Kafka Value**: The value schema is inferred based on the connector’s
  examination of records in firebase database.
* **Kafka Header**: The configured database reference will be a part of Kafka
  record header with key as `Database_Reference`.