<a id="cc-object-storage-backup-restore"></a>

# Back Up and Restore Data with Object Storage Connectors in Confluent Cloud

The fully managed Google Cloud Storage (GCS) sink and source connectors can back up complete Apache Kafka® records to
object storage and restore them to a Kafka topic. A matched pair of connector modes makes this possible:

- `BACKUP_FULL_RECORD` on the sink connector
- `RESTORE_BACKUP_FULL_RECORD` on the source connector

In these modes, the sink connector wraps the key, value, headers, and record metadata (topic, partition, offset, and
timestamp) in a single self-describing envelope record and writes it to object storage. The sink connector also
writes the original schemas to object storage as separate files. The source connector reads the envelope
records and the schema files, and produces records to Kafka with the same key, value, headers, and metadata, and with the
original schema registered in the destination Schema Registry.

This guide describes how the backup and restore modes work, the fidelity you can expect, and how to configure
the connectors.

<a id="cc-object-storage-backup-restore-supported-connectors"></a>

## Supported connectors

The following fully managed connectors support backup and restore.

### Sink connectors

Use this connector to back up a topic:

- [Google Cloud Storage Sink connector](cc-gcs-sink.md#cc-gcs-connect-sink)

### Source connectors

Use this connector to restore a topic:

- [Google Cloud Storage Source connector](cc-gcs-source.md#cc-gcs-connect-source)

<a id="cc-object-storage-backup-restore-modes"></a>

## Connector modes

Each connector has a `mode` configuration property that sets the operating mode.

| Connector   | Valid values                            | Default   |
|-------------|-----------------------------------------|-----------|
| Sink        | `GENERIC`, `BACKUP_FULL_RECORD`         | `GENERIC` |
| Source      | `GENERIC`, `RESTORE_BACKUP_FULL_RECORD` | `GENERIC` |

`GENERIC` is the default. It’s how the connectors behaved before the backup and restore modes existed, so existing
connectors aren’t affected. Use the full-record modes when you need to restore more than record values. In
`GENERIC` mode:

- The sink connector writes record values as the primary data, and the source connector has no way to restore keys or
  headers.
- The topic, partition, offset, and timestamp are encoded in the object path and file name rather than stored as
  fields of the record.
- When the source connector reads data back, the Connect framework adds metadata such as `connect.name`
  and `connect.version` to the schema, so the schema differs from the original.

The full-record modes close these gaps by storing every record facet in one envelope and by backing up the
original schemas separately.

#### IMPORTANT
**Mode can’t be changed on an existing connector.** Switching `mode` on a running source connector produces
undefined behavior because offset state from the previous mode might cause the connector to read files in the
wrong format. Treat the mode as immutable. To change the mode, delete the connector and create a new one with
a different name.

### When to use backup and restore

Backup and restore with object storage connectors isn’t a replacement for [Cluster Linking](../multi-cloud/cluster-linking/index.md#cloud-cluster-linking)
or a dedicated disaster recovery solution. Cluster Linking provides real-time, low-latency replication with offset
preservation, and is the recommended approach for disaster recovery with low recovery point and recovery time
objectives, active-passive failover, and multi-cluster real-time mirroring.

Use the backup and restore modes for scenarios where Cluster Linking doesn’t apply:

- **Long-term archival with replay:** Store months or years of data in object storage and replay it into a topic on demand.
- **Cross-environment data seeding:** Populate development or staging environments from production backups.
- **Compliance and audit:** Keep an immutable backup of Kafka data with full record fidelity.
- **Schema Registry migration:** Restore data with the original schemas to a new Schema Registry instance.
- **Cold storage rehydration:** Bring archived data back to Kafka for reprocessing after the original retention period
  has expired.
- **Multi-cloud or cross-region restore:** Back up from one cloud provider or region and restore to another.

<a id="cc-object-storage-backup-restore-workflow"></a>

## Backup and restore steps

At a high level, you back up a topic with a sink connector and restore it with a source connector.

### Back up a topic

1. On the fully managed GCS Sink connector, set `mode` to `BACKUP_FULL_RECORD`.
2. If you plan to restore the backup, use the `DefaultPartitioner`. The source connector
   restores only the files that this partitioner writes. For more information, see
   [Partitioners and ordering](#cc-object-storage-backup-restore-partitioners).
3. The connector wraps every Kafka record in an envelope that contains the key, value, headers, topic, partition,
   offset, timestamp, and schema metadata.
4. The connector backs up the original schemas as `.avsc` or `.json` files under the
   `_metadata/schemas/` prefix in the destination bucket.
5. The connector writes the envelope records to object storage in the format you set in `output.data.format`:
   `AVRO`, `JSON`, or `PARQUET`. If you plan to restore with the GCS Source connector, use `AVRO` or `JSON`,
   because that connector can’t read Parquet files.

### Restore a topic

1. Create the destination topic.

   #### IMPORTANT
   The destination topic must have at least as many partitions as the source topic. If it has fewer partitions,
   the connector task stays in the `RUNNING` state but produces no records, or drops records, without an error
   or a warning. For more information, see [Source connector limitations](#cc-object-storage-backup-restore-limits-source).
2. On the fully managed GCS Source connector, set `mode` to `RESTORE_BACKUP_FULL_RECORD` and point the connector
   at the backup bucket path.
3. The connector reads the envelope records and extracts the original key, value, and headers.
4. The connector serializes the data by using the backed-up schema, without the metadata that the
   Connect framework adds.
5. The connector produces the records to the destination topic with the same schema, key, value, headers, and
   partition as the original. It can also preserve the original timestamp. For more information, see
   [Timestamp preservation and retention](#cc-object-storage-backup-restore-timestamps).

<a id="cc-object-storage-backup-restore-fidelity"></a>

## Fidelity

You choose one of two fidelity levels for the record data (key, value, and headers) through the converter
configuration of the connector.

| Level                           | How to achieve it                                                                                                    | Data guarantee                                                                                                                                                                  |
|---------------------------------|----------------------------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| **Semantic Fidelity** (default) | Use typed converters that match the original producer: Avro, JSON Schema, JSON, String, or<br/>primitive converters. | Every field value round-trips correctly. Wire bytes can differ when the format allows ordering freedom,<br/>such as maps, or when primitive serializers emit variant encodings. |
| **Byte Fidelity**               | Set `ByteArrayConverter` for the key, value, and header converters.                                                  | The exact Kafka wire bytes are preserved end to end. The connector doesn’t interact with Schema Registry.                                                                       |

#### IMPORTANT
**Only Semantic Fidelity backs up schemas.** Byte Fidelity bypasses Schema Registry entirely, so the sink connector doesn’t
back up schema files and the source connector doesn’t register schemas in the destination Schema Registry. For Byte Fidelity,
the destination Schema Registry must already contain the same schemas under the same IDs. If it doesn’t, the connector still
produces the records, but downstream consumers fail to resolve the schemas. For more information, see
[Byte Fidelity](#cc-object-storage-backup-restore-byte-fidelity).

Both levels use the same `BACKUP_FULL_RECORD` and `RESTORE_BACKUP_FULL_RECORD` modes, the same envelope layout,
and the same tombstone behavior. The only difference is the choice of `key.converter`, `value.converter`, and
`header.converter`. You can choose a different fidelity level for each connector, but a single connector can’t mix
typed converters and `ByteArrayConverter` on a per-record basis.

### Semantic Fidelity

Use the converters that your producers already use: `AvroConverter`, `JsonSchemaConverter`,
`JsonConverter` (with or without `schemas.enable`), `StringConverter`, or any primitive converter.
`ProtobufConverter` isn’t supported at this time.

With Semantic Fidelity:

- The backup files are typed and self-describing, so you can read them directly with analytical tools such
  as Amazon Athena, Google BigQuery, Snowflake, Trino, and Apache Spark without using the source connector.
- Restored values are byte-identical for Schema Registry-backed converters (Avro and JSON Schema) when the schema
  contains only order-preserving constructs, such as records, arrays, primitives, and enums.
- Restored values are semantically equal, but not necessarily byte-identical, when the schema contains maps or
  other constructs whose wire ordering the format doesn’t specify, such as Avro maps. The same is true when
  `output.data.format` is `PARQUET`, because the connector writes Parquet files through Avro. The table later in
  this section lists the sources of divergence.
- Restored values are semantically equal for converters that don’t use Schema Registry, such as `JsonConverter` and primitive
  converters. Field ordering, whitespace, and numeric widening can differ, but every field value round-trips
  correctly. Primitive converters usually produce identical bytes, but identity isn’t guaranteed for all data.
- Headers of primitive and `BYTES` types are typed. Headers of `MAP`, `ARRAY`, and `STRUCT` types are
  restored as `STRING`. For more information, see [Header handling](#cc-object-storage-backup-restore-headers).

Use Semantic Fidelity when the backup also serves analytical use cases, or when downstream tools expect readable,
typed files.

Even when all converter settings are correct, Semantic Fidelity can produce wire bytes that differ from the original
producer. All of these differences are semantic no-ops, because parsers accept both forms and the field values
compare equal.

| Source of divergence                    | Example                                                                                                                                                       | Impact                       |
|-----------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------|
| Map iteration order                     | A `map<string,int32>` is serialized with entries in a different order, because Java `HashMap` iteration<br/>isn’t deterministic.                              | One to three bytes of drift. |
| Float and double encoding variants      | The same numeric value has multiple valid IEEE-754 encodings, such as NaN payloads or +0 and -0.                                                              | Rare.                        |
| JSON serialization order and whitespace | `output.data.format = JSON` writes object properties in the order of the Connect `Struct` fields,<br/>which can differ from the original producer’s encoding. | None on field values.        |
| String encoding variations              | Non-ASCII UTF-8 canonicalization.                                                                                                                             | None on field values.        |

<a id="cc-object-storage-backup-restore-byte-fidelity"></a>

### Byte Fidelity

Set `key.converter`, `value.converter`, and `header.converter` to
`org.apache.kafka.connect.converters.ByteArrayConverter`. The connector treats every payload as opaque bytes
on both backup and restore.

With Byte Fidelity:

- Restore reproduces the exact wire bytes of the key, the value, and every header, regardless of the original
  serializer, including formats that Schema Registry doesn’t support, such as MessagePack, Thrift, or a proprietary binary protocol.
- The connector doesn’t interact with Schema Registry on backup or restore. Schema backup, the subject naming strategy,
  reference resolution, and registration in the destination Schema Registry are all bypassed.
- It’s the most defensive option when the requirement is that the bytes never change, such as audit archives,
  regulatory holds, and cryptographically signed payloads.

The trade-offs are:

- The backup files contain raw serialized bytes with no schema information. No analytical tool can query them
  without external knowledge of the source format.
- If downstream consumers resolve schemas by ID, you must migrate the schemas to the destination Schema Registry separately,
  with a method that preserves schema IDs. For example, use [Schema Linking](../sr/schema-linking.md#schema-linking) or Schema Registry
  `IMPORT` mode pre-registration.

Use Byte Fidelity when the only goal of the backup is bit-perfect archival and opaque files are acceptable.

| Concern                                             | Semantic Fidelity                               | Byte Fidelity               |
|-----------------------------------------------------|-------------------------------------------------|-----------------------------|
| Query backup files with analytical tools            | Yes                                             | No (opaque bytes)           |
| Wire-byte identity on restore                       | Only for order-preserving schemas               | Always, for any format      |
| Schema Registry dependency                          | Required on both sides                          | Not required on either side |
| Schema migration to the destination Schema Registry | Automatic (the connector registers the schemas) | Manual                      |
| Formats that Schema Registry doesn’t support        | No                                              | Yes                         |
| Operational complexity                              | Lower                                           | Higher                      |

### Schema fidelity

With Semantic Fidelity, the sink connector backs up each original Avro or JSON Schema as a separate file
next to the data. On restore, the source connector registers that original schema in the destination Schema Registry, so the
schema definition, including field names, types, structure, documentation, and defaults, is identical to the original.
The connector doesn’t add `connect.name` or `connect.version` metadata.

Schema Registry assigns schema IDs and version numbers sequentially when schemas are registered, so the destination
Schema Registry assigns different IDs and versions than the source. Consumers that resolve schemas by subject name, which is the
recommended pattern, aren’t affected.

#### NOTE
The backup doesn’t include Schema Registry governance features, such as encryption keys, client-side field level
encryption (CSFLE) rules, metadata, or migration rules. For more information, see
[General limitations](#cc-object-storage-backup-restore-limits-generic).

<a id="cc-object-storage-backup-restore-headers"></a>

### Header handling

Kafka headers pass through the configured `header.converter` on both the sink and source connectors, and the
converter you choose determines the fidelity of the headers. You can set `header.converter` as an additional
configuration property.

For strict header fidelity, set `header.converter` to
`org.apache.kafka.connect.converters.ByteArrayConverter` on both the sink and source connectors. The connector
then treats every header as opaque bytes, and downstream consumers receive `BYTES`-typed headers that they must
deserialize. Keep the default `SimpleHeaderConverter` only if all headers are plain UTF-8 text or simple
primitives, and no consumer depends on typed header APIs.

The default `SimpleHeaderConverter` is lossy in the following cases:

- **High-precision floats:** Precision is lost in the `FLOAT32` round trip.
- **JSON-like text with whitespace:** The converter parses the text as a `MAP` and writes it back in compact form,
  so whitespace is lost.
- **Raw binary (non-UTF-8 bytes):** The UTF-8 decode replaces invalid bytes with `U+FFFD`, so the header is corrupted.
- **Struct, map, and array headers:** The connector restores these as `STRING` values that contain JSON-like text.

`ByteArrayConverter` on both sides is byte-identical in all of these cases.

<a id="cc-object-storage-backup-restore-subjects"></a>

### Subject naming strategy

Schema Registry tracks compatibility, evolution, and reference resolution by subject name, not by schema content. A single
schema ID can be registered under several subjects, so the subject is a part of the backup metadata. The
[subject name strategy](../sr/fundamentals/serdes-develop/index.md#sr-schemas-subject-name-strategy) configured on the producer determines how a schema is
named, and the source connector must apply a compatible strategy to register schemas under the correct subjects in the
destination Schema Registry.

The source connector supports all three strategies.

| Strategy                      | Example subject           | Derived from                                     |
|-------------------------------|---------------------------|--------------------------------------------------|
| `TopicNameStrategy` (default) | `orders-value`            | Topic name plus a `-key` or `-value` suffix.     |
| `RecordNameStrategy`          | `com.example.User`        | Fully qualified record name in the schema.       |
| `TopicRecordNameStrategy`     | `orders-com.example.User` | Topic name plus the fully qualified record name. |

**How backup captures the subject:** The converter computes the subject at backup time with the configured
`value.subject.name.strategy` (and `key.subject.name.strategy` for keys). The sink connector doesn’t look up the
subject of a schema ID in Schema Registry, because several subjects can share one ID. The subject that the producer used is
stored in the backup metadata with the schema definition.

**How restore applies the subject:** The behavior depends on the role of the schema.

| Schema role      | Subject on restore                                                                                                                                                                                                         |
|------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Main schema      | The destination Schema Registry converter recomputes the subject from the destination topic name and the<br/>`value.subject.name.strategy` of the source connector. The subject respects `restore.topic.prefix`.           |
| Reference schema | The subject is preserved as is from the backup metadata. The raw text of the main schema refers to its<br/>dependencies by their original subject names, so renaming a reference subject would invalidate the main schema. |

**Interaction with the topic prefix:** The following examples use `restore.topic.prefix = "restored-"` with a
source topic named `orders`.

| Strategy                  | Source subject            | Destination topic   | Destination main schema subject    |
|---------------------------|---------------------------|---------------------|------------------------------------|
| `TopicNameStrategy`       | `orders-value`            | `restored-orders`   | `restored-orders-value`            |
| `RecordNameStrategy`      | `com.example.User`        | `restored-orders`   | `com.example.User` (unchanged)     |
| `TopicRecordNameStrategy` | `orders-com.example.User` | `restored-orders`   | `restored-orders-com.example.User` |

#### WARNING
Set `value.subject.name.strategy` (and `key.subject.name.strategy` if keys use Schema Registry) on the source connector
to match the strategy of the original producer. If the strategies don’t match, schemas are registered under the
wrong subjects in the destination Schema Registry, and the connector doesn’t raise an error. For more information, see
[Source connector limitations](#cc-object-storage-backup-restore-limits-source).

<a id="cc-object-storage-backup-restore-behavior"></a>

## Behavior

This section describes how backup and restore operate at runtime.

### Delivery semantics

- **Backup sink connector:** Exactly-once when the connector uses a deterministic partitioner, such as
  `DefaultPartitioner`, which is the existing behavior of the connector. The connector deduplicates object storage
  writes by using upload sequence numbers, so a task failure and restart doesn’t produce duplicate envelope files or
  duplicate records within a file.
- **Restore source connector:** At-least-once. After a task failure and restart, the connector might produce records
  to the destination topic again. Downstream consumers should be idempotent or tolerate duplicates.

### File rotation

The sink connector rotation settings, `rotate.schedule.interval.ms`, `flush.size`, and `rotate.interval.ms`,
determine the size and time span of each envelope file. The source connector reads whole files, so rotation
granularity sets the smallest unit that you can replay. Smaller files give finer replay granularity but create more
objects to manage. Larger files create fewer objects but give coarser replay units.

<a id="cc-object-storage-backup-restore-file-names"></a>

### File name offset padding

In `BACKUP_FULL_RECORD` mode, the sink connector sets `filename.offset.zero.pad.width` to `20` instead of the
default `10`. The connector ignores any value that you set for this property. The source connector reads backup files
in lexicographic order by file name, and Kafka offsets are 64-bit numbers with up to 19 digits. A pad width of `10`
breaks the ordering after the offsets in a partition exceed 10 digits, which is about 10 billion records, and the
source connector then reads the files out of order. A pad width of `20` covers the full offset range.

As a result, every backup file name contains a 20-digit, zero-padded offset, for example
`topic+0+00000000000000000075.avro`. You don’t need to take any action.

### Read scope

The source connector reads every file under the configured `topics.dir` prefix in the source bucket. The connector
can’t select files by offset range, timestamp, or file name pattern, and `topics.regex.list` doesn’t apply in
restore mode. As a result:

- Isolate each backup set in its own bucket or prefix. If you mix multiple backup generations under the same prefix,
  the connector replays all of them.
- To replay a subset of files, copy the files to a new prefix and point the source connector at that prefix.

<a id="cc-object-storage-backup-restore-layout"></a>

### Directory layout

Each source connector is bound to one `topics.dir` and one converter configuration. A connector has exactly one
`key.converter`, one `value.converter`, one `header.converter`, and one storage format for the files that it
reads. The connector deserializes every object under its `topics.dir` with those converters and that format, and it
can’t select converters for each topic or each object.

Because of this, choose the `topics.dir` layout when you deploy the sink connector. The layout that you choose at
backup time determines what you can restore later. To change it afterward, you must run the backup again or move the
objects in the bucket.

#### Group topics by converter and format

All topics that you back up into the same `topics.dir` must use the same converters and the same storage format. If a
`topics.dir` contains topics that were produced with different converters, such as some topics with `AvroConverter`
and some with `ByteArrayConverter`, or topics that were written in different storage formats, the source connector
fails. Depending on which record the connector reads first, the task stops with a converter `DataException`. For
example, restoring with `ByteArrayConverter` from a directory that also contains structured records fails with the
following error:

```text
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
Caused by: org.apache.kafka.connect.errors.DataException:
Invalid schema type for ByteArrayConverter: STRUCT
    at org.apache.kafka.connect.converters.ByteArrayConverter.fromConnectData(...)
```

To avoid this error, use a dedicated `topics.dir` for each combination of converters and storage format, and restore
each directory with a separate source connector that has the matching converter and format configuration. For example,
you can set `topics.dir` to `backups/bytes`, `backups/avro`, and `backups/json-schema` for the following
layout:

```text
<bucket>/
  backups/
    bytes/            # ByteArrayConverter topics
      topic-a/
      topic-b/
    avro/             # AvroConverter topics, AVRO storage format
      topic-c/
      topic-d/
    json-schema/      # JsonSchemaConverter topics
      topic-e/
```

#### Split large topic sets across directories

Even when all topics use the same converters and storage format, putting a large number of topics under one
`topics.dir` causes throughput and operational problems on restore. A large number is in the low hundreds or more.

- The source connector lists and schedules work for every topic under the directory, which increases startup and
  listing time.
- The `tasks.max` setting of one connector limits the work, which caps horizontal parallelism.
- You can’t restore a subset of the topics, because the directory is the smallest restore unit. The whole set is tied
  to the lifecycle of one connector.

For a large number of topics, split the topics across several `topics.dir` groups, so that an independent source
connector can restore each group in parallel:

```text
<bucket>/
  backups/
    avro/
      group-01/       # subset of the AvroConverter topics
      group-02/       # subset of the AvroConverter topics
      group-03/       # subset of the AvroConverter topics
```

As a starting point, keep each `topics.dir` to tens of topics instead of hundreds, unless benchmarking in your
environment justifies a larger group.

#### Directory layout checklist

1. Decide the `topics.dir` layout when you deploy the sink connector. The layout on the sink side determines what
   you can restore later.
2. Use one set of converters and one storage format for each `topics.dir`. This is a hard requirement.
3. For a large number of topics, split them across several `topics.dir` groups to enable parallel restore.
4. Keep a manifest outside the bucket that records which topics are under each `topics.dir` and which converters and
   storage format apply. With a manifest, restore operators don’t have to infer them.
5. Before you start a restore, check that the converters and storage format of the source connector match every topic
   under the selected directory.

<a id="cc-object-storage-backup-restore-partitioners"></a>

### Partitioners and ordering

The sink connector in `BACKUP_FULL_RECORD` mode supports `DefaultPartitioner`, `TimeBasedPartitioner`,
`DailyPartitioner`, and `HourlyPartitioner`. It doesn’t support `FieldPartitioner`.

The source connector doesn’t have a partitioner setting. In `RESTORE_BACKUP_FULL_RECORD` mode, it reads only the
files that a sink connector wrote with `DefaultPartitioner`, and it can’t restore files that the time-based
partitioners wrote. To restore a backup, set `partitioner.class` to `DefaultPartitioner` on the sink connector
when you create the backup.

`DefaultPartitioner` writes each Kafka partition to its own directory, so the directory structure encodes the original
partition. This is how per-partition ordering is preserved end to end across a backup and restore cycle.

### Error handling

Fail fast on both sides of a backup and restore cycle. A record that the connector tolerates silently becomes an
invisible gap in the restored stream.

- **Sink connector:** Set `errors.tolerance` to `none`. If the connector hits a serialization failure, an envelope
  wrap error, or a non-retriable storage write error, the task fails instead of silently dropping the record. A backup
  is only useful if it’s complete. If you route bad records to a dead letter queue (DLQ) at backup time, those records
  aren’t in the backup files, so you can’t restore them. A failed task shows the problem while the record is still in
  the source topic.
- **Source connector:** Keep `behavior.on.error` set to `FAIL` (the default) so the task stops at the first
  unprocessable envelope file. Valid values are `FAIL` and `IGNORE`, and `IGNORE` risks silent data loss when the connector skips a corrupted
  file. Also set `errors.tolerance` to `none` so that framework-level record errors, such as deserialization,
  converter, and Schema Registry registration errors, fail the task immediately.

### Schema evolution

When new schema versions appear in the source topic during the backup window, the sink connector writes new schema
files next to the data files. The source connector registers each distinct schema version in the destination Schema Registry in
the order it encounters them.

The compatibility setting for the subject in the destination Schema Registry must allow all of the schema versions. If the
destination compatibility setting is stricter than the source, schema registration during restore can fail. Check the
compatibility setting of the destination Schema Registry before you deploy the source connector.

<a id="cc-object-storage-backup-restore-timestamps"></a>

### Timestamp preservation and retention

The `restore.preserve.timestamps` property on the source connector controls the timestamp of restored records.

| Setting           | Behavior                                                                                                                                                                      |
|-------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `false` (default) | Restored records get the wall-clock time of the restore. The destination topic looks like a new stream, and<br/>the normal `retention.ms` applies from the moment of restore. |
| `true`            | Restored records keep the original producer timestamps from the backup.                                                                                                       |

#### WARNING
If you set `restore.preserve.timestamps` to `true` and the backup records are older than the `retention.ms`
of the destination topic, the broker deletes the restored records almost immediately because they appear expired.
Restoring a backup that is several months old into a topic with seven-day retention produces an empty topic.

Before you enable preserved timestamps:

1. Review the `retention.ms` and `retention.bytes` settings of the destination topic.
2. Choose one of the following options:
   - Disable time-based retention on the destination topic by setting `retention.ms` to `-1`.
   - Set `retention.ms` to a window that covers the age span of the backup plus the time your downstream consumers
     need to process the restored records.
   - Keep `restore.preserve.timestamps` set to `false` so that the connector stamps records at restore time and
     normal retention applies.
3. Run the restore.
4. Optionally, reset `retention.ms` to its production value after the downstream consumers process the restored
   records.

### Restore producer compression

If you don’t set `producer.override.compression.type` on the source connector, the connector produces records to the
destination topic uncompressed, regardless of the codec of the original producer. For a backup that was originally
produced with Snappy, GZIP, or Zstd, the destination topic can grow much larger than the source topic.

#### IMPORTANT
Set `producer.override.compression.type` as an additional configuration property on the source connector to
match the codec of the original producer. Confluent Cloud doesn’t let you configure the compression type on the topic, so you
must set it on the restore producer.

<a id="cc-object-storage-backup-restore-configs"></a>

## Configuration

In Confluent Cloud, you select storage and message formats through the connector configuration fields instead of setting
the underlying Connect configuration properties, such as `format.class`. The connector maps your selections to
the underlying properties.

The following sections describe the configuration properties that are relevant to backup and restore. For the
complete list of properties for each connector, see the configuration properties of the
[GCS Sink connector](cc-gcs-sink.md#cc-gcs-storage-sink-config-properties) and the
[GCS Source connector](cc-gcs-source.md#cc-gcs-source-config-properties). The connector documentation is the source of truth for
property names, defaults, and valid values.

#### NOTE
`ProtobufConverter` isn’t supported in `BACKUP_FULL_RECORD` and `RESTORE_BACKUP_FULL_RECORD` modes at
this time. This includes the `PROTOBUF` format selection, which uses `ProtobufConverter`. For more
information, see [Format-specific limitations](#cc-object-storage-backup-restore-limits-format).

### Sink connector properties

| Property             | Default                | Description                                                                                                                                                                                                                                                                  |
|----------------------|------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `mode`               | `GENERIC`              | The operating mode of the connector. Valid values are `GENERIC` and `BACKUP_FULL_RECORD`.                                                                                                                                                                                    |
| `input.data.format`  | None                   | The format of the Kafka record values that the connector reads from the topic. This selection drives the value<br/>converter. Valid values are `AVRO`, `BYTES`, `JSON`, `JSON_SR`, `PROTOBUF`, and `STRING`. Selecting<br/>`BYTES` puts the value on the Byte Fidelity path. |
| `input.key.format`   | `JSON`                 | The format of the Kafka record keys that the connector reads from the topic. This selection drives the key<br/>converter. Valid values are `AVRO`, `BYTES`, `JSON`, `JSON_SR`, `PROTOBUF`, and `STRING`. Selecting<br/>`BYTES` puts the key on the Byte Fidelity path.       |
| `output.data.format` | `${input.data.format}` | The format of the envelope files that the connector writes to object storage. Valid values are `AVRO`,<br/>`JSON`, and `PARQUET` in `BACKUP_FULL_RECORD` mode.                                                                                                               |

In `BACKUP_FULL_RECORD` mode:

- `output.data.format` must be `AVRO`, `JSON`, or `PARQUET`. The connector rejects `BYTES` at deployment.
- To keep the exact key bytes, set `input.key.format` to `BYTES`. The key schema then isn’t backed up, so if the keys
  use a schema, you must migrate it with an ID-preserving method. For more information, see
  [Byte Fidelity limitations](#cc-object-storage-backup-restore-limits-byte).

### Source connector properties

| Property                      | Default                | Description                                                                                                                                                                                                                                                                                                                                                        |
|-------------------------------|------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `mode`                        | `GENERIC`              | The operating mode of the connector. Valid values are `GENERIC` and `RESTORE_BACKUP_FULL_RECORD`.                                                                                                                                                                                                                                                                  |
| `input.data.format`           | None                   | The format of the backup files that the connector reads from object storage. In<br/>`RESTORE_BACKUP_FULL_RECORD` mode, valid values are `AVRO` and `JSON`, and the format must match<br/>`output.data.format` on the sink connector. This applies to both Semantic Fidelity and Byte Fidelity backups.                                                             |
| `output.data.format`          | `${input.data.format}` | The format of the records that the connector produces to Kafka. Valid values are `AVRO`, `BYTES`, `JSON`,<br/>`JSON_SR`, `PROTOBUF`, and `STRING`. Match the value format of the original producer. `BYTES` produces<br/>raw bytes to Kafka without registering a schema in the destination Schema Registry.                                                       |
| `output.key.format`           | `STRING`               | The format of the record keys that the connector produces to Kafka. Valid values are `AVRO`, `BYTES`, `JSON`,<br/>`JSON_SR`, `PROTOBUF`, and `STRING`. Match the key format of the original producer. `BYTES` is available<br/>only when `mode` is `RESTORE_BACKUP_FULL_RECORD` and produces raw key bytes without registering a key schema.                       |
| `restore.topic.prefix`        | Empty                  | A prefix that the connector adds to the original topic name on restore. Available only when `mode` is<br/>`RESTORE_BACKUP_FULL_RECORD`.                                                                                                                                                                                                                            |
| `restore.preserve.timestamps` | `false`                | When `true`, restored records keep the original Kafka timestamps from the backup. When `false`, the connector<br/>stamps records with the time of the restore. Review `retention.ms` of the destination topic before you enable<br/>this property. For more information, see [Timestamp preservation and retention](#cc-object-storage-backup-restore-timestamps). |

In `RESTORE_BACKUP_FULL_RECORD` mode:

- `input.data.format` must match `output.data.format` on the sink connector, so that both sides agree on the
  envelope container format.
- `output.data.format` and `output.key.format` should match the value and key formats of the original producer.
- `PARQUET` isn’t supported as `input.data.format` on the GCS Source connector. For more information, see
  [Source connector limitations](#cc-object-storage-backup-restore-limits-source).

### Choose the sink output format

Choose `output.data.format` based on the converter of the source topic.

| Converter of the source topic                               | Recommended `output.data.format`                                                                                                |
|-------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------|
| `AvroConverter`                                             | `AVRO` (default). Don’t use `PARQUET` if you plan to restore with the GCS Source connector, which can’t read<br/>Parquet files. |
| `JsonSchemaConverter`                                       | `JSON`                                                                                                                          |
| `JsonConverter` (`schemas.enable` set to `false` or `true`) | `JSON` (required)                                                                                                               |
| `ByteArrayConverter` (Byte Fidelity)                        | `AVRO` or `JSON`                                                                                                                |
| `StringConverter` and primitive converters                  | `JSON`                                                                                                                          |

#### `JSON` output format

With `output.data.format = JSON`, the sink connector writes one JSON line for each envelope record. The connector
uses an internal Kafka `JsonConverter`, and the `format.json.schema.enable` property controls whether that
converter writes the schema.

| Property                    | Default   | Description                                                                                                                                                                                                    |
|-----------------------------|-----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `format.json.schema.enable` | `false`   | Must be `true` in backup mode. If it’s `false`, the envelope schema is lost on write, restore isn’t<br/>possible, and the connector rejects the configuration. Applies to both the sink and source connectors. |

<a id="cc-object-storage-backup-restore-converter-configs"></a>

### Converter configuration

Set the following converter properties symmetrically on both the sink and source connectors. The connector rejects
a deployment that is missing a required property. If your keys use the same converter type as your values, set the
same properties with the `key.converter.` prefix.

#### Avro

| Property                                       | Default   | Recommendation                                                                                                                         |
|------------------------------------------------|-----------|----------------------------------------------------------------------------------------------------------------------------------------|
| `value.converter.enhanced.avro.schema.support` | `false`   | Set to `true` on both sides to preserve enums. Without it, enums are flattened to strings and lose their<br/>enum identity on restore. |

#### JSON Schema

| Property                                         | Default   | Recommendation                                                                                                                                                                                                                                                                                |
|--------------------------------------------------|-----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `value.converter.preserve.additional.properties` | `false`   | Set to `true` on both sides if your schemas contain `additionalProperties` fields whose extra entries<br/>must survive restore. If you set it on only one side, the behavior is lossy. The legacy<br/>`connect.type: map` schema annotation is still supported for fields that are pure maps. |
| `value.converter.use.optional.for.nonrequired`   | `false`   | Set to `true` on both sides if producers might omit non-required fields entirely. For more information, see<br/>[Format-specific limitations](#cc-object-storage-backup-restore-limits-format).                                                                                               |

#### JSON (schemaless)

| Property                                    | Default                                | Recommendation                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
|---------------------------------------------|----------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `value.converter.schemas.enable`            | `true` (`false` for the key converter) | Must match on the sink and source connectors, either both `true` or both `false`. A mismatch changes the<br/>shape of the data without an error. If the sink connector uses `true` and the source connector uses<br/>`false`, the restored records lose the envelope wrapper. If the sink connector uses `false` and the<br/>source connector uses `true`, the restored records are wrapped in `{"schema": null, "payload": {...}}`.<br/>The sink connector, source connector, and producer must all agree on the format. |
| `value.converter.replace.null.with.default` | `true`                                 | Set to `false` on both sides. The default replaces null values with the schema default on both<br/>deserialization and serialization, which changes null fields in the restored records. For example, a schema<br/>default of `"UNSPECIFIED"` and a null wire value restore as `"UNSPECIFIED"`.                                                                                                                                                                                                                           |

Without a schema (`schemas.enable` set to `false`), the converter infers Connect types from the JSON types:
`INT8`, `INT16`, and `INT32` become `INT64`, `FLOAT32` and `FLOAT64` become `FLOAT64`, a `BYTES` value
with the Decimal logical type becomes `STRING`, and `STRUCT` becomes a `MAP` with `STRING` keys. The value types
in the map follow the JSON type of each field, so nested objects become nested maps. If `JsonConverter` is also the
only consumer of the restored data, the data is semantically equal end to end. If you convert the restored data to Avro
or JSON Schema downstream, the type coercion is visible and lossy.

#### Key converter properties

The connectors expose the following `key.converter.*` properties so that you can tune key converters in the same way
as value converters. Each property has the same meaning as its `value.converter.*` counterpart.

- `key.converter.enhanced.avro.schema.support`
- `key.converter.replace.null.with.default`
- `key.converter.use.optional.for.nonrequired`
- `key.converter.preserve.additional.properties`
- `key.converter.object.additional.properties`

### Minimum configuration by fidelity level

#### Semantic Fidelity

Semantic Fidelity is the only path that backs up schemas. Use the following recommended configuration for each
producer format, and apply it under both `value.converter.*` and `key.converter.*` when keys use the same
converter type.

| Producer format                 | Recommended configuration                                                                                                                                                                                                                                           |
|---------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Avro                            | `value.converter.enhanced.avro.schema.support=true` on both sides.                                                                                                                                                                                                  |
| JSON Schema                     | `value.converter.preserve.additional.properties=true` if `additionalProperties` fields exist and their<br/>extra entries must survive. `value.converter.use.optional.for.nonrequired=true` if producers might omit<br/>non-required fields. Set both on both sides. |
| JSON (schemaless)               | `value.converter.schemas.enable` matches on both sides, and `value.converter.replace.null.with.default=false`<br/>on both sides.                                                                                                                                    |
| String and primitive converters | None. The converters work with default settings.                                                                                                                                                                                                                    |

#### Byte Fidelity

Apply all of the following settings on both the sink and source connectors.

| Slot              | Property                                       | Value                                                                                                                                                                                                                                                                                                                  |
|-------------------|------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Value             | `value.converter`                              | `org.apache.kafka.connect.converters.ByteArrayConverter`. In the Console, you can select `BYTES` for<br/>`input.data.format` on the sink connector and `output.data.format` on the source connector.                                                                                                                   |
| Key               | `key.converter`                                | `org.apache.kafka.connect.converters.ByteArrayConverter`. In the Console, you can select `BYTES` for<br/>`input.key.format` on the sink connector and `output.key.format` on the source connector.                                                                                                                     |
| Headers           | `header.converter`                             | `org.apache.kafka.connect.converters.ByteArrayConverter`. The Console doesn’t provide a selection for this<br/>property, so set it as an additional configuration property.                                                                                                                                            |
| Storage container | `output.data.format` on the sink connector     | `AVRO` or `JSON`. `BYTES` isn’t a valid storage container.                                                                                                                                                                                                                                                             |
| Topic prefix      | `restore.topic.prefix` on the source connector | Allowed, but you must pre-register schemas in the destination Schema Registry under the subject that matches the renamed<br/>topic. For more information, see [Byte Fidelity limitations](#cc-object-storage-backup-restore-limits-byte).                                                                              |
| Prerequisite      | Schema migration that preserves IDs            | Required. Byte Fidelity doesn’t back up any schemas. Before you run a backup and restore, use<br/>[Schema Linking](../sr/schema-linking.md#schema-linking), Schema Registry `IMPORT` mode pre-registration, or an equivalent method to put<br/>matching schemas in the destination Schema Registry under the same IDs. |

You can also apply Byte Fidelity to a single slot while the other slots use typed converters. For example, a common
pattern is to set `header.converter` to `ByteArrayConverter` for byte-exact headers while the value and key use
typed converters.

| Mix                              | Settings                                                                                           | Consequence                                                                                                                                                                                                                           |
|----------------------------------|----------------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Byte Fidelity for the value only | `input.data.format=BYTES` (sink), `output.data.format=BYTES` (source), with typed key and headers. | The value schema isn’t backed up, so you must migrate the value schema with an ID-preserving method. The key<br/>and headers use the typed flow.                                                                                      |
| Byte Fidelity for the key only   | `input.key.format=BYTES` (sink), `output.key.format=BYTES` (source), with typed value and headers. | The key schema isn’t backed up, so you must migrate the key schema with an ID-preserving method.                                                                                                                                      |
| Byte Fidelity for headers only   | `header.converter=ByteArrayConverter` on both sides, with typed value and key.                     | Headers carry raw bytes end to end and have no Schema Registry implication. Struct, map, and array header values<br/>round-trip without loss. For more information, see [Header handling](#cc-object-storage-backup-restore-headers). |

### Header-based schema IDs

Records produced with `HeaderSchemaIdSerializer` store the schema ID in the Kafka record headers
`__value_schema_id` and `__key_schema_id`, instead of in the Schema Registry magic-byte prefix of the message.

- **Backup:** The sink connector detects header-based IDs automatically. It checks the headers first and falls back to
  the prefix of the value. You don’t need to change the sink connector configuration.
- **Restore:** You must configure the source connector to write the header-based format. Otherwise, the connector adds
  the standard Schema Registry magic-byte prefix on top of the preserved header, which breaks byte fidelity. Set the following
  properties on the source connector:
  ```text
  value.converter.value.schema.id.serializer=io.confluent.kafka.serializers.schema.id.HeaderSchemaIdSerializer

  # If keys were also produced with header-based IDs:
  key.converter.key.schema.id.serializer=io.confluent.kafka.serializers.schema.id.HeaderSchemaIdSerializer
  ```

  Connect removes one `value.converter.` prefix before it passes the configuration to the converter. The
  `value.schema.id.serializer` property still needs the `value.converter.` prefix in the connector
  configuration so that it survives that step. The same applies to keys.

#### NOTE
The message viewer in Confluent Cloud Console uses the standard prefix-based deserializer, so the first record of a
topic with header-based IDs might appear as raw bytes while later records appear normally. The record data is
intact. This is a limitation of the viewer only.

<a id="cc-object-storage-backup-restore-limits"></a>

## Limitations

The following limitations apply to the backup and restore modes.

<a id="cc-object-storage-backup-restore-limits-generic"></a>

### General limitations

These limitations apply to both the sink and source connectors.

- **Single message transforms (SMTs) aren’t supported.** Deployment fails if the `transforms` property is set. The
  connector wraps and unwraps the envelope inside its own processing path. An SMT before the wrap would store mutated
  data in the backup, and an SMT after the unwrap would change the restored records. Both defeat the fidelity
  guarantee. There’s no workaround. Run any transformation on a downstream consumer or in a separate pipeline that
  reads from the restored topic.
- **Bytes as the output format isn’t supported.** Deployment fails if `output.data.format` is `BYTES` on the sink
  connector, because the envelope can’t be encoded as a raw byte array without a container structure. Use `AVRO`
  or `JSON`. You can also use `PARQUET` if you don’t plan to restore with the GCS Source connector.
- **FieldPartitioner isn’t supported.** For a successful deployment, ensure that your
  connector uses the `DefaultPartitioner`. For more information,
  see [Partitioners and ordering](#cc-object-storage-backup-restore-partitioners).
- **The schema backup doesn’t include Schema Registry governance features.** The schema files include the record
  schema, but not key encryption keys (KEKs), data encryption keys (DEKs), CSFLE rules, schema metadata, domain
  rules, or migration rules. On restore, the connector can’t decrypt encrypted fields, and ciphertext round-trips
  as is. Reapply the Schema Registry governance configuration to the destination Schema Registry separately, before downstream consumers
  process the restored stream.
- **Compacted topics need validation.** The connectors preserve tombstones (records with a null value) through the
  envelope and restore them as they were. For heavily compacted topics, where compaction can race with a backup,
  validate the end-to-end behavior for your retention and compaction settings before you rely on backup and restore
  for recovery.

### Sink connector limitations

**Schema files are mandatory.** Each unique schema version that the sink connector sees produces a schema file under
the `_metadata/schemas/` prefix. You can’t turn off schema file writes in backup mode, because the source connector
relies on them to rebuild schemas. Expect a small amount of extra storage for each schema version compared to a sink
connector that doesn’t use backup mode.

<a id="cc-object-storage-backup-restore-limits-source"></a>

### Source connector limitations

- **All topics in a directory must share the same converters and storage format.** A source connector has one key
  converter, one value converter, one header converter, and one storage format, and it applies them to every object
  under its `topics.dir`. If the directory contains topics with different converters or storage formats, the task
  fails with a converter error. Back up each combination into its own `topics.dir`, and restore each directory with
  a separate source connector. For more information, see [Directory layout](#cc-object-storage-backup-restore-layout).
- **Parquet isn’t supported on the GCS Source connector.** Deployment fails if `input.data.format` is `PARQUET`,
  because the Parquet format isn’t available on this connector. If you wrote a backup as `PARQUET` on GCS, you can’t
  restore it with the fully managed source connector. If you need to restore, back up with `AVRO` or `JSON`.
- **The destination topic must exist with a matching partition count.** If the destination topic has fewer partitions
  than the source topic, the task reports `RUNNING` but produces no records, or drops records without an error,
  because the records for the missing partitions have nowhere to go. Create the destination topic with at least as
  many partitions as the source topic before you deploy the source connector. Extra partitions are allowed and stay
  unused.
- **A subject naming strategy mismatch causes silent misdirection.** If `value.subject.name.strategy` (or
  `key.subject.name.strategy`) on the source connector doesn’t match the strategy of the original producer,
  the connector registers schemas under the wrong subjects without an error. Consumers of the restored topic then can’t
  resolve schemas by the subject name that they expect, and compatibility rules might not match what the producer
  intended. Before you deploy the source connector:
  1. Identify the `value.subject.name.strategy` (and `key.subject.name.strategy` if keys use Schema Registry) of the producer.
  2. Set the same strategy on the source connector as an additional configuration property.
  3. Check that the subject in the backup metadata matches the subject that the restore configuration produces.

  For more information, see [Subject naming strategy](#cc-object-storage-backup-restore-subjects).
- **Schema IDs in the destination Schema Registry differ from the source.** The connector preserves the schema
  definition, but the destination Schema Registry assigns its own IDs and version numbers. Consumers that resolve schemas by
  subject name aren’t affected. Consumers that depend on specific schema IDs don’t find their original IDs. If you
  need to preserve schema IDs, use Byte Fidelity with an ID-preserving schema migration.
- **Legacy backup files aren’t readable.** The task fails with schema deserialization errors if the object storage
  path contains output from a sink connector that wasn’t in `BACKUP_FULL_RECORD` mode, such as plain Avro or JSON
  records without the envelope. `RESTORE_BACKUP_FULL_RECORD` expects every data file to be an envelope that a
  `BACKUP_FULL_RECORD` sink connector wrote.
- **Only files that \`\`DefaultPartitioner\`\` wrote are readable.** The source connector has no partitioner setting, and
  it reads only the directory structure that `DefaultPartitioner` creates, with one directory for each Kafka
  partition. It can’t restore files that `TimeBasedPartitioner`, `DailyPartitioner`, or `HourlyPartitioner`
  wrote. Set `partitioner.class` to `DefaultPartitioner` on the sink connector for every backup that you plan to
  restore. For more information, see [Partitioners and ordering](#cc-object-storage-backup-restore-partitioners).
- **The mode can’t be changed on an existing connector.** For more information, see
  [Connector modes](#cc-object-storage-backup-restore-modes).
- **The file metadata headers property is ignored.** The `file.metadata.headers.enable` property has no effect in
  `RESTORE_BACKUP_FULL_RECORD` mode. In `GENERIC` mode, the property adds file metadata, such as the file name,
  path, last modified time, and size, to each record as headers. In restore mode, the envelope already contains the
  original Kafka headers, so the connector always restores the original producer headers from the envelope. The
  connector accepts the property at deployment, and setting it to `true` doesn’t change the behavior.

<a id="cc-object-storage-backup-restore-limits-format"></a>

### Format-specific limitations

**Avro:** Enums are flattened to strings and lose their enum identity on restore unless you set
`value.converter.enhanced.avro.schema.support=true` on both the sink and source connectors. If the keys are Avro and
need to keep enum identity, set `key.converter.enhanced.avro.schema.support=true` as well. For more information, see
[Converter configuration](#cc-object-storage-backup-restore-converter-configs).

**Protobuf:** `ProtobufConverter` isn’t supported in backup and restore modes at this time.

**JSON Schema**

- **Omitted non-required fields cause a sink task failure:** The sink task fails with a “null used for required field”
  error if a record omits a non-required field, because `JsonSchemaConverter` doesn’t carry the JSON Schema
  `required: false` setting to the Connect schema as `optional=true`. Use one of the following workarounds, in
  order of preference:
  1. Declare the field as explicitly nullable in the JSON Schema, for example `{"type": ["number", "null"]}`.
  2. Always include all fields in the produced messages, including the non-required ones.
  3. Set `value.converter.use.optional.for.nonrequired=true` on both sides. The sink connector then accepts records
     with omitted fields, but the restored topic contains explicit `"field": null` entries for the fields that the
     original producer omitted. Consumers that strictly validate against the JSON Schema might see validation errors,
     such as `expected Number, found Null`, on the restored topic that didn’t occur on the original topic.
- **additionalProperties values are dropped by default:** To preserve the values of `additionalProperties` fields
  across restore, set `value.converter.preserve.additional.properties=true` on both the sink and source connectors.
  If you set it on only one side, the behavior stays lossy. The default behavior, which drops these values, is
  kept for backward compatibility. For fields that are pure maps, you can instead add the `"connect.type": "map"`
  annotation to the field in the schema, or set `output.data.format` to `JSON`, which avoids the Avro map conversion.

<a id="cc-object-storage-backup-restore-limits-byte"></a>

### Byte Fidelity limitations

Byte Fidelity uses `ByteArrayConverter` for the key, value, and headers. This is a converter choice, and it’s
separate from the storage format limitation that `output.data.format` can’t be `BYTES`.

- **The destination Schema Registry must already contain the matching schema IDs.** The sink connector preserves the
  original schema IDs inside the backed-up wire bytes, but those IDs mean nothing in the destination Schema Registry unless the
  schemas are there under the same IDs. The connector doesn’t migrate schemas for Byte Fidelity, and automatic
  migration isn’t supported. If the IDs don’t match, the connector still produces the records unchanged, because it
  doesn’t interact with Schema Registry in this mode, but downstream consumers fail to resolve the schemas. Before you run a
  Byte Fidelity backup and restore, use one of the following methods:
  - [Schema Linking](../sr/schema-linking.md#schema-linking) from the source Schema Registry to the destination Schema Registry.
  - Schema Registry `IMPORT` mode pre-registration, where you manually register each required schema at its original ID in the
    destination Schema Registry.
  - An equivalent out-of-band migration that preserves the original schema IDs.
- **A topic prefix changes the subject that consumers look for.** The connector doesn’t block `restore.topic.prefix`
  for Byte Fidelity. However, the prefix renames the destination topic, and topic-derived subject naming strategies,
  such as `TopicNameStrategy`, derive the subject from the renamed topic. If you pre-registered the schemas under
  the original topic’s subject, which is the typical case, consumers of the renamed topic can’t find the schema. To
  avoid this, pre-register the schemas under the subjects that match the renamed topic, leave `restore.topic.prefix`
  empty and rely on the original subject, or use Semantic Fidelity, which registers the schema under the subject that the
  restored topic resolves to.

<a id="cc-object-storage-backup-restore-related"></a>

## Related content

- [Google Cloud Storage Sink connector](cc-gcs-sink.md#cc-gcs-connect-sink)
- [Google Cloud Storage Source connector](cc-gcs-source.md#cc-gcs-connect-source)
- [Cluster Linking](../multi-cloud/cluster-linking/index.md#cloud-cluster-linking)
- [Schema Linking](../sr/schema-linking.md#schema-linking)
- [Google Cloud Storage Source connector backup and restore for Confluent Platform](https://docs.confluent.io/kafka-connectors/gcs-source/current/backup-and-restore/overview.html)