From 8338116dffe847931cae1212333af04338ea1d45 Mon Sep 17 00:00:00 2001 From: Thiago Nunes Date: Thu, 18 Mar 2021 15:58:42 +1100 Subject: [PATCH] feat!: customer-managed encryption keys for Spanner (#666) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add support for encrypted databases * fix: fix deps and clirr failures * tests: add additional tests for keys * tests: remove IT and add unit * fix: set null instead of default instance * fix: does not set encryption info if null Does not set encryption info in the request if it is null * fix: fixes dependencies * feature: adds support for encrypted backup Adds the possibility to set encryption config info in the creation of a backup. * feature: adds support for restoring encrypted dbs * Revert "tests: remove IT and add unit" This reverts commit cc19cf2efd32007ecd351c3b0c1b5942256c31ce. * fix: makes the setEncryptionConfigInfo public This is so a backup can be encrypted * feature: adds tests for cmek Adds tests for creating encrypted database, creating encrypted backups and restoring encrypted databases. * fix: removes keys after test finishes Destroy keys used in CMEK tests * fix: fixes clirr errors * fix: ignores failing cmek tests Ignores the failing CMEK tests until the backend support is enabled in production. * fix: uses wrapper encryption info for backups * fix: fixes clirr issues * fix: re-orders clirr issues * fix: addresses PR comments * test: fixes database admin client tests * chore: re-formats the code * chore: fixes clirr checks * tests: adds unit tests for domain classes Adds unit tests for EncryptionConfigInfo, EncryptionConfig, Backup and Restore. * chore: renames EncryptionConfigInfo Renames EncryptionConfigInfo to EncryptionConfig in order to mirror what is the protobuf definition. * tests: do not create a key on CMEK test Instead use an existing key and fails if the key is not present. * feat: allows multiple encryption configs Allows customer managed encryption for create databases (google default encryption is just nullifying the value here). Allows customer managed encryption, google default encryption and database encryption for create backups. Allows customer managed encryption, google default encryption and backup encryption for restore databases. * docs: adds java doc to Restore class * chore: refactors pom.xml Uses variables to define project id and instance id for running integration tests. * test: fixes cmek integration test * chore: fixes linting * Revert "chore: refactors pom.xml" This reverts commit d182b8316bf78322f22dcc7d48d6955cecd844f7. * test: unifies cmek backup and restore tests * chore: adds toString to encryption classes * docs: updates DatabaseInfo javadoc Co-authored-by: Knut Olav Løite * docs: updates Restore javadocs Co-authored-by: Knut Olav Løite * docs: updates DatabaseInfo javadocs Co-authored-by: Knut Olav Løite * fix: addresses PR comments * tests: reformats Co-authored-by: Olav Loite --- .../clirr-ignored-differences.xml | 71 +++++++- google-cloud-spanner/pom.xml | 3 +- .../java/com/google/cloud/spanner/Backup.java | 6 +- .../com/google/cloud/spanner/BackupInfo.java | 75 ++++++++- .../com/google/cloud/spanner/Database.java | 2 + .../cloud/spanner/DatabaseAdminClient.java | 94 ++++++++++- .../spanner/DatabaseAdminClientImpl.java | 70 +++++--- .../google/cloud/spanner/DatabaseInfo.java | 50 +++++- .../com/google/cloud/spanner/Restore.java | 109 ++++++++++++ .../encryption/BackupEncryptionConfig.java | 23 +++ .../encryption/CustomerManagedEncryption.java | 66 ++++++++ .../EncryptionConfigProtoMapper.java | 77 +++++++++ .../spanner/encryption/EncryptionConfigs.java | 45 +++++ .../spanner/encryption/EncryptionInfo.java | 92 +++++++++++ .../encryption/GoogleDefaultEncryption.java | 30 ++++ .../encryption/RestoreEncryptionConfig.java | 23 +++ .../encryption/UseBackupEncryption.java | 30 ++++ .../encryption/UseDatabaseEncryption.java | 33 ++++ .../cloud/spanner/spi/v1/GapicSpannerRpc.java | 125 ++++++++------ .../cloud/spanner/spi/v1/SpannerRpc.java | 24 +-- .../com/google/cloud/spanner/BackupTest.java | 56 +++---- .../spanner/DatabaseAdminClientImplTest.java | 155 ++++++++++++++++-- .../google/cloud/spanner/DatabaseTest.java | 71 ++++++++ .../com/google/cloud/spanner/RestoreTest.java | 54 ++++++ .../CustomerManagedEncryptionTest.java | 56 +++++++ .../EncryptionConfigProtoMapperTest.java | 139 ++++++++++++++++ .../encryption/EncryptionConfigsTest.java | 55 +++++++ .../encryption/EncryptionInfoTest.java | 79 +++++++++ .../google/cloud/spanner/it/ITBackupTest.java | 66 ++++++-- 29 files changed, 1614 insertions(+), 165 deletions(-) create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/Restore.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/BackupEncryptionConfig.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/CustomerManagedEncryption.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapper.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigs.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionInfo.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/GoogleDefaultEncryption.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/RestoreEncryptionConfig.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseBackupEncryption.java create mode 100644 google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseDatabaseEncryption.java create mode 100644 google-cloud-spanner/src/test/java/com/google/cloud/spanner/RestoreTest.java create mode 100644 google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/CustomerManagedEncryptionTest.java create mode 100644 google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapperTest.java create mode 100644 google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigsTest.java create mode 100644 google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionInfoTest.java diff --git a/google-cloud-spanner/clirr-ignored-differences.xml b/google-cloud-spanner/clirr-ignored-differences.xml index 98dd288b28..f340f5e196 100644 --- a/google-cloud-spanner/clirr-ignored-differences.xml +++ b/google-cloud-spanner/clirr-ignored-differences.xml @@ -319,7 +319,7 @@ com/google/cloud/spanner/Value java.util.List getNumericArray() - + 7012 @@ -406,7 +406,7 @@ com/google/cloud/spanner/AbstractLazyInitializer java.lang.Object initialize() - + 7004 @@ -504,4 +504,71 @@ com/google/cloud/spanner/DatabaseAdminClient com.google.api.gax.longrunning.OperationFuture createBackup(com.google.cloud.spanner.Backup) + + + + 7004 + com/google/cloud/spanner/spi/v1/SpannerRpc + com.google.api.gax.longrunning.OperationFuture createDatabase(java.lang.String, java.lang.String, java.lang.Iterable) + + + 7004 + com/google/cloud/spanner/spi/v1/SpannerRpc + com.google.api.gax.longrunning.OperationFuture createBackup(java.lang.String, java.lang.String, com.google.spanner.admin.database.v1.Backup) + + + 7004 + com/google/cloud/spanner/spi/v1/SpannerRpc + com.google.api.gax.longrunning.OperationFuture restoreDatabase(java.lang.String, java.lang.String, java.lang.String) + + + 7004 + com/google/cloud/spanner/spi/v1/GapicSpannerRpc + com.google.api.gax.longrunning.OperationFuture createDatabase(java.lang.String, java.lang.String, java.lang.Iterable) + + + 7004 + com/google/cloud/spanner/spi/v1/GapicSpannerRpc + com.google.api.gax.longrunning.OperationFuture createBackup(java.lang.String, java.lang.String, com.google.spanner.admin.database.v1.Backup) + + + 7004 + com/google/cloud/spanner/spi/v1/GapicSpannerRpc + com.google.api.gax.longrunning.OperationFuture restoreDatabase(java.lang.String, java.lang.String, java.lang.String) + + + 7012 + com/google/cloud/spanner/DatabaseAdminClient + com.google.api.gax.longrunning.OperationFuture createDatabase(com.google.cloud.spanner.Database, java.lang.Iterable) + + + 7012 + com/google/cloud/spanner/DatabaseAdminClient + com.google.api.gax.longrunning.OperationFuture createBackup(com.google.cloud.spanner.Backup) + + + 7012 + com/google/cloud/spanner/DatabaseAdminClient + com.google.api.gax.longrunning.OperationFuture restoreDatabase(com.google.cloud.spanner.Restore) + + + 7012 + com/google/cloud/spanner/DatabaseAdminClient + com.google.cloud.spanner.Database$Builder newDatabaseBuilder(com.google.cloud.spanner.DatabaseId) + + + 7012 + com/google/cloud/spanner/DatabaseAdminClient + com.google.cloud.spanner.Restore$Builder newRestoreBuilder(com.google.cloud.spanner.BackupId, com.google.cloud.spanner.DatabaseId) + + + 7013 + com/google/cloud/spanner/DatabaseInfo$Builder + com.google.cloud.spanner.DatabaseInfo$Builder setEncryptionConfig(com.google.cloud.spanner.encryption.CustomerManagedEncryption) + + + 7013 + com/google/cloud/spanner/BackupInfo$Builder + com.google.cloud.spanner.BackupInfo$Builder setEncryptionConfig(com.google.cloud.spanner.encryption.BackupEncryptionConfig) + diff --git a/google-cloud-spanner/pom.xml b/google-cloud-spanner/pom.xml index 1cd5665190..4534486af9 100644 --- a/google-cloud-spanner/pom.xml +++ b/google-cloud-spanner/pom.xml @@ -73,6 +73,7 @@ com.google.cloud.spanner.GceTestEnvConfig projects/gcloud-devel/instances/spanner-testing gcloud-devel + projects/gcloud-devel/locations/us-central1/keyRings/spanner-test-keyring/cryptoKeys/spanner-test-key 3000 @@ -383,7 +384,7 @@ - generate-test-sql-scripts diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Backup.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Backup.java index a0052cd381..27308c0400 100644 --- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Backup.java +++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Backup.java @@ -23,6 +23,7 @@ import com.google.api.gax.paging.Page; import com.google.cloud.Policy; import com.google.cloud.Timestamp; +import com.google.cloud.spanner.encryption.EncryptionInfo; import com.google.longrunning.Operation; import com.google.spanner.admin.database.v1.CreateBackupMetadata; import com.google.spanner.admin.database.v1.RestoreDatabaseMetadata; @@ -61,10 +62,6 @@ public Backup build() { /** Creates a backup on the server based on the source of this {@link Backup} instance. */ public OperationFuture create() { - Preconditions.checkState( - getExpireTime() != null, "Cannot create a backup without an expire time"); - Preconditions.checkState( - getDatabase() != null, "Cannot create a backup without a source database"); return dbClient.createBackup(this); } @@ -184,6 +181,7 @@ static Backup fromProto( .setExpireTime(Timestamp.fromProto(proto.getExpireTime())) .setVersionTime(Timestamp.fromProto(proto.getVersionTime())) .setDatabase(DatabaseId.of(proto.getDatabase())) + .setEncryptionInfo(EncryptionInfo.fromProtoOrNull(proto.getEncryptionInfo())) .setProto(proto) .build(); } diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/BackupInfo.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/BackupInfo.java index 199e6ae2ae..0657ff2b7b 100644 --- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/BackupInfo.java +++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/BackupInfo.java @@ -18,6 +18,8 @@ import com.google.api.client.util.Preconditions; import com.google.cloud.Timestamp; +import com.google.cloud.spanner.encryption.BackupEncryptionConfig; +import com.google.cloud.spanner.encryption.EncryptionInfo; import com.google.spanner.admin.database.v1.Database; import java.util.Objects; import javax.annotation.Nullable; @@ -29,8 +31,29 @@ public abstract static class Builder { abstract Builder setSize(long size); + /** + * Returned when retrieving a backup. + * + *

The encryption information for the backup. If the encryption key protecting this resource + * is customer managed, then kms_key_version will be filled. + */ + abstract Builder setEncryptionInfo(EncryptionInfo encryptionInfo); + abstract Builder setProto(com.google.spanner.admin.database.v1.Backup proto); + /** + * Optional for creating a new backup. + * + *

The encryption configuration to be used for the backup. The possible configurations are + * {@link com.google.cloud.spanner.encryption.CustomerManagedEncryption}, {@link + * com.google.cloud.spanner.encryption.GoogleDefaultEncryption} and {@link + * com.google.cloud.spanner.encryption.UseDatabaseEncryption}. + * + *

If no encryption config is given the backup will be created with the same encryption as + * set by the database ({@link com.google.cloud.spanner.encryption.UseDatabaseEncryption}). + */ + public abstract Builder setEncryptionConfig(BackupEncryptionConfig encryptionConfig); + /** * Required for creating a new backup. * @@ -70,6 +93,8 @@ abstract static class BuilderImpl extends Builder { private Timestamp versionTime; private DatabaseId database; private long size; + private BackupEncryptionConfig encryptionConfig; + private EncryptionInfo encryptionInfo; private com.google.spanner.admin.database.v1.Backup proto; BuilderImpl(BackupId id) { @@ -83,6 +108,8 @@ abstract static class BuilderImpl extends Builder { this.versionTime = other.versionTime; this.database = other.database; this.size = other.size; + this.encryptionConfig = other.encryptionConfig; + this.encryptionInfo = other.encryptionInfo; this.proto = other.proto; } @@ -113,12 +140,24 @@ public Builder setDatabase(DatabaseId database) { return this; } + @Override + public Builder setEncryptionConfig(BackupEncryptionConfig encryptionConfig) { + this.encryptionConfig = encryptionConfig; + return this; + } + @Override Builder setSize(long size) { this.size = size; return this; } + @Override + Builder setEncryptionInfo(EncryptionInfo encryptionInfo) { + this.encryptionInfo = encryptionInfo; + return this; + } + @Override Builder setProto(@Nullable com.google.spanner.admin.database.v1.Backup proto) { this.proto = proto; @@ -142,12 +181,16 @@ public enum State { private final Timestamp versionTime; private final DatabaseId database; private final long size; + private final BackupEncryptionConfig encryptionConfig; + private final EncryptionInfo encryptionInfo; private final com.google.spanner.admin.database.v1.Backup proto; BackupInfo(BuilderImpl builder) { this.id = builder.id; this.state = builder.state; this.size = builder.size; + this.encryptionConfig = builder.encryptionConfig; + this.encryptionInfo = builder.encryptionInfo; this.expireTime = builder.expireTime; this.versionTime = builder.versionTime; this.database = builder.database; @@ -174,6 +217,22 @@ public long getSize() { return size; } + /** + * Returns the {@link BackupEncryptionConfig} to encrypt the backup during its creation. Returns + * null if no customer-managed encryption key should be used. + */ + public BackupEncryptionConfig getEncryptionConfig() { + return encryptionConfig; + } + + /** + * Returns the {@link EncryptionInfo} of the backup if the backup is encrypted, or null + * if this backup is not encrypted. + */ + public EncryptionInfo getEncryptionInfo() { + return encryptionInfo; + } + /** Returns the expire time of the backup. */ public Timestamp getExpireTime() { return expireTime; @@ -206,6 +265,8 @@ public boolean equals(Object o) { return id.equals(that.id) && state == that.state && size == that.size + && Objects.equals(encryptionConfig, that.encryptionConfig) + && Objects.equals(encryptionInfo, that.encryptionInfo) && Objects.equals(expireTime, that.expireTime) && Objects.equals(versionTime, that.versionTime) && Objects.equals(database, that.database); @@ -213,13 +274,21 @@ public boolean equals(Object o) { @Override public int hashCode() { - return Objects.hash(id, state, size, expireTime, versionTime, database); + return Objects.hash( + id, state, size, encryptionConfig, encryptionInfo, expireTime, versionTime, database); } @Override public String toString() { return String.format( - "Backup[%s, %s, %d, %s, %s, %s]", - id.getName(), state, size, expireTime, versionTime, database); + "Backup[%s, %s, %d, %s, %s, %s, %s, %s]", + id.getName(), + state, + size, + encryptionConfig, + encryptionInfo, + expireTime, + versionTime, + database); } } diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Database.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Database.java index a442ad2399..05ba3f2edf 100644 --- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Database.java +++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Database.java @@ -22,6 +22,7 @@ import com.google.api.gax.paging.Page; import com.google.cloud.Policy; import com.google.cloud.Timestamp; +import com.google.cloud.spanner.encryption.CustomerManagedEncryption; import com.google.common.base.Preconditions; import com.google.longrunning.Operation; import com.google.spanner.admin.database.v1.CreateBackupMetadata; @@ -185,6 +186,7 @@ static Database fromProto( .setRestoreInfo(RestoreInfo.fromProtoOrNullIfDefaultInstance(proto.getRestoreInfo())) .setVersionRetentionPeriod(proto.getVersionRetentionPeriod()) .setEarliestVersionTime(Timestamp.fromProto(proto.getEarliestVersionTime())) + .setEncryptionConfig(CustomerManagedEncryption.fromProtoOrNull(proto.getEncryptionConfig())) .setProto(proto) .build(); } diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClient.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClient.java index eae1a3cdf4..2e4fd09951 100644 --- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClient.java +++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClient.java @@ -24,6 +24,7 @@ import com.google.longrunning.Operation; import com.google.spanner.admin.database.v1.CreateBackupMetadata; import com.google.spanner.admin.database.v1.CreateDatabaseMetadata; +import com.google.spanner.admin.database.v1.CreateDatabaseRequest; import com.google.spanner.admin.database.v1.RestoreDatabaseMetadata; import com.google.spanner.admin.database.v1.UpdateDatabaseDdlMetadata; import java.util.List; @@ -68,9 +69,53 @@ public interface DatabaseAdminClient { OperationFuture createDatabase( String instanceId, String databaseId, Iterable statements) throws SpannerException; + /** + * Creates a database in a Cloud Spanner instance. Any configuration options in the {@link + * Database} instance will be included in the {@link CreateDatabaseRequest}. + * + *

Example to create an encrypted database. + * + * 

{@code
+   * Database dbInfo =
+   *     dbClient
+   *         .newDatabaseBuilder(DatabaseId.of("my-project", "my-instance", "my-database"))
+   *         .setEncryptionConfig(
+   *             EncryptionConfig.ofKey(
+   *                 "projects/my-project/locations/some-location/keyRings/my-keyring/cryptoKeys/my-key"))
+   *         .build();
+   * Operation op = dbAdminClient
+   *     .createDatabase(
+   *         dbInfo,
+   *         Arrays.asList(
+   *             "CREATE TABLE Singers (\n"
+   *                 + "  SingerId   INT64 NOT NULL,\n"
+   *                 + "  FirstName  STRING(1024),\n"
+   *                 + "  LastName   STRING(1024),\n"
+   *                 + "  SingerInfo BYTES(MAX)\n"
+   *                 + ") PRIMARY KEY (SingerId)",
+   *             "CREATE TABLE Albums (\n"
+   *                 + "  SingerId     INT64 NOT NULL,\n"
+   *                 + "  AlbumId      INT64 NOT NULL,\n"
+   *                 + "  AlbumTitle   STRING(MAX)\n"
+   *                 + ") PRIMARY KEY (SingerId, AlbumId),\n"
+   *                 + "  INTERLEAVE IN PARENT Singers ON DELETE CASCADE"));
+   * Database db = op.waitFor().getResult();
+   * }
+ * + * @see also #createDatabase(String, String, Iterable) + */ + OperationFuture createDatabase( + Database database, Iterable statements) throws SpannerException; + + /** Returns a builder for a {@code Database} object with the given id. */ + Database.Builder newDatabaseBuilder(DatabaseId id); + /** Returns a builder for a {@code Backup} object with the given id. */ Backup.Builder newBackupBuilder(BackupId id); + /** Returns a builder for a {@link Restore} object with the given source and destination */ + Restore.Builder newRestoreBuilder(BackupId source, DatabaseId destination); + /** * Creates a new backup from a database in a Cloud Spanner instance. * @@ -90,8 +135,8 @@ OperationFuture createDatabase( * Backup backup = op.get(); * } * - * @param instanceId the id of the instance where the database to backup is located and where the - * backup will be created. + * @param sourceInstanceId the id of the instance where the database to backup is located and + * where the backup will be created. * @param backupId the id of the backup which will be created. It must conform to the regular * expression [a-z][a-z0-9_\-]*[a-z0-9] and be between 2 and 60 characters in length. * @param databaseId the id of the database to backup. @@ -102,21 +147,27 @@ OperationFuture createBackup( throws SpannerException; /** - * Creates a new backup from a database in a Cloud Spanner instance. + * Creates a new backup from a database in a Cloud Spanner. Any configuration options in the + * {@link Backup} instance will be included in the {@link + * com.google.spanner.admin.database.v1.CreateBackupRequest}. * - *

Example to create a backup. + *

Example to create an encrypted backup. * * 

{@code
-   * BackupId backupId     = BackupId.of("project", "instance", "backup-id");
+   * BackupId backupId = BackupId.of("project", "instance", "backup-id");
    * DatabaseId databaseId = DatabaseId.of("project", "instance", "database-id");
-   * Timestamp expireTime  = Timestamp.ofTimeMicroseconds(expireTimeMicros);
+   * Timestamp expireTime = Timestamp.ofTimeMicroseconds(expireTimeMicros);
    * Timestamp versionTime = Timestamp.ofTimeMicroseconds(versionTimeMicros);
+   * EncryptionConfig encryptionConfig =
+   *         EncryptionConfig.ofKey(
+   *             "projects/my-project/locations/some-location/keyRings/my-keyring/cryptoKeys/my-key"));
    *
    * Backup backupToCreate = dbAdminClient
    *     .newBackupBuilder(backupId)
    *     .setDatabase(databaseId)
    *     .setExpireTime(expireTime)
    *     .setVersionTime(versionTime)
+   *     .setEncryptionConfig(encryptionConfig)
    *     .build();
    *
    * OperationFuture op = dbAdminClient.createBackup(backupToCreate);
@@ -138,7 +189,7 @@ OperationFuture createBackup(
    * String backupId           = my_backup_id;
    * String restoreInstanceId  = my_db_instance_id;
    * String restoreDatabaseId  = my_database_id;
-   * OperationFuture op = dbAdminClient
+   * OperationFuture op = dbAdminClient
    *     .restoreDatabase(
    *         backupInstanceId,
    *         backupId,
@@ -153,10 +204,37 @@ OperationFuture createBackup(
    *     be a different instance than where the backup is stored.
    * @param restoreDatabaseId the id of the database to restore to.
    */
-  public OperationFuture restoreDatabase(
+  OperationFuture restoreDatabase(
       String backupInstanceId, String backupId, String restoreInstanceId, String restoreDatabaseId)
       throws SpannerException;
 
+  /**
+   * Restore a database from a backup. The database that is restored will be created and may not
+   * already exist.
+   *
+   * 
Example to restore an encrypted database.
+   *
+   * 
{@code
+   * final Restore restore = dbAdminClient
+   *     .newRestoreBuilder(
+   *         BackupId.of("my-project", "my-instance", "my-backup"),
+   *         DatabaseId.of("my-project", "my-instance", "my-database")
+   *     )
+   *     .setEncryptionConfig(EncryptionConfig.ofKey(
+   *         "projects/my-project/locations/some-location/keyRings/my-keyring/cryptoKeys/my-key"))
+   *     .build();
+   *
+   * final OperationFuture op = dbAdminClient
+   *     .restoreDatabase(restore);
+   *
+   * Database database = op.get();
+   * }

+   *
+   * @param restore a {@link Restore} instance with the backup source and destination database
+   */
+  OperationFuture restoreDatabase(Restore restore)
+      throws SpannerException;
+
   /** Lists long-running database operations on the specified instance. */
   Page listDatabaseOperations(String instanceId, ListOption... options);
 
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClientImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClientImpl.java
index a5eed214a4..6129e0fa2d 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClientImpl.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseAdminClientImpl.java
@@ -25,6 +25,7 @@
 import com.google.cloud.Policy;
 import com.google.cloud.Policy.DefaultMarshaller;
 import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.DatabaseInfo.State;
 import com.google.cloud.spanner.Options.ListOption;
 import com.google.cloud.spanner.SpannerImpl.PageFetcher;
 import com.google.cloud.spanner.spi.v1.SpannerRpc;
@@ -72,21 +73,37 @@ private static String randomOperationId() {
     return ("r" + uuid.toString()).replace("-", "_");
   }
 
+  @Override
+  public Database.Builder newDatabaseBuilder(DatabaseId databaseId) {
+    return new Database.Builder(this, databaseId);
+  }
+
   @Override
   public Backup.Builder newBackupBuilder(BackupId backupId) {
     return new Backup.Builder(this, backupId);
   }
 
+  @Override
+  public Restore.Builder newRestoreBuilder(BackupId source, DatabaseId destination) {
+    return new Restore.Builder(source, destination);
+  }
+
   @Override
   public OperationFuture restoreDatabase(
       String backupInstanceId, String backupId, String restoreInstanceId, String restoreDatabaseId)
       throws SpannerException {
-    String databaseInstanceName = getInstanceName(restoreInstanceId);
-    String backupName = getBackupName(backupInstanceId, backupId);
+    return restoreDatabase(
+        newRestoreBuilder(
+                BackupId.of(projectId, backupInstanceId, backupId),
+                DatabaseId.of(projectId, restoreInstanceId, restoreDatabaseId))
+            .build());
+  }
 
-    OperationFuture
-        rawOperationFuture =
-            rpc.restoreDatabase(databaseInstanceName, restoreDatabaseId, backupName);
+  @Override
+  public OperationFuture restoreDatabase(Restore restore)
+      throws SpannerException {
+    final OperationFuture
+        rawOperationFuture = rpc.restoreDatabase(restore);
 
     return new OperationFutureImpl(
         rawOperationFuture.getPollingFuture(),
@@ -114,29 +131,25 @@ public Database apply(Exception e) {
   public OperationFuture createBackup(
       String instanceId, String backupId, String databaseId, Timestamp expireTime)
       throws SpannerException {
-    final Backup backup =
+    final Backup backupInfo =
         newBackupBuilder(BackupId.of(projectId, instanceId, backupId))
             .setDatabase(DatabaseId.of(projectId, instanceId, databaseId))
             .setExpireTime(expireTime)
             .build();
-    return createBackup(backup);
+
+    return createBackup(backupInfo);
   }
 
   @Override
-  public OperationFuture createBackup(final Backup backup) {
-    final String instanceId = backup.getInstanceId().getInstance();
-    final String databaseId = backup.getDatabase().getDatabase();
-    final String backupId = backup.getId().getBackup();
-    final com.google.spanner.admin.database.v1.Backup.Builder backupBuilder =
-        com.google.spanner.admin.database.v1.Backup.newBuilder()
-            .setDatabase(getDatabaseName(instanceId, databaseId))
-            .setExpireTime(backup.getExpireTime().toProto());
-    if (backup.getVersionTime() != null) {
-      backupBuilder.setVersionTime(backup.getVersionTime().toProto());
-    }
-    final String instanceName = getInstanceName(instanceId);
+  public OperationFuture createBackup(Backup backupInfo)
+      throws SpannerException {
+    Preconditions.checkArgument(
+        backupInfo.getExpireTime() != null, "Cannot create a backup without an expire time");
+    Preconditions.checkArgument(
+        backupInfo.getDatabase() != null, "Cannot create a backup without a source database");
+
     final OperationFuture
-        rawOperationFuture = rpc.createBackup(instanceName, backupId, backupBuilder.build());
+        rawOperationFuture = rpc.createBackup(backupInfo);
 
     return new OperationFutureImpl<>(
         rawOperationFuture.getPollingFuture(),
@@ -154,6 +167,7 @@ public Backup apply(OperationSnapshot snapshot) {
                     .setExpireTime(proto.getExpireTime())
                     .setVersionTime(proto.getVersionTime())
                     .setState(proto.getState())
+                    .setEncryptionInfo(proto.getEncryptionInfo())
                     .build(),
                 DatabaseAdminClientImpl.this);
           }
@@ -281,11 +295,19 @@ public Backup fromProto(com.google.spanner.admin.database.v1.Backup proto) {
   @Override
   public OperationFuture createDatabase(
       String instanceId, String databaseId, Iterable statements) throws SpannerException {
-    // CreateDatabase() is not idempotent, so we're not retrying this request.
-    String instanceName = getInstanceName(instanceId);
-    String createStatement = "CREATE DATABASE `" + databaseId + "`";
+    return createDatabase(
+        new Database(DatabaseId.of(projectId, instanceId, databaseId), State.UNSPECIFIED, this),
+        statements);
+  }
+
+  @Override
+  public OperationFuture createDatabase(
+      Database database, Iterable statements) throws SpannerException {
+    String createStatement = "CREATE DATABASE `" + database.getId().getDatabase() + "`";
     OperationFuture
-        rawOperationFuture = rpc.createDatabase(instanceName, createStatement, statements);
+        rawOperationFuture =
+            rpc.createDatabase(
+                database.getId().getInstanceId().getName(), createStatement, statements, database);
     return new OperationFutureImpl(
         rawOperationFuture.getPollingFuture(),
         rawOperationFuture.getInitialFuture(),
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseInfo.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseInfo.java
index 5ba9f0aa76..101fdd4e64 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseInfo.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseInfo.java
@@ -17,6 +17,7 @@
 package com.google.cloud.spanner;
 
 import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.encryption.CustomerManagedEncryption;
 import com.google.common.base.Preconditions;
 import java.util.Objects;
 import javax.annotation.Nullable;
@@ -34,6 +35,15 @@ public abstract static class Builder {
 
     abstract Builder setEarliestVersionTime(Timestamp earliestVersionTime);
 
+    /**
+     * Optional for creating a new backup.
+     *
+     * 
The encryption configuration to be used for the database. The only encryption, other than
+     * Google's default encryption, is a customer managed encryption with a provided key. If no
+     * encryption is provided, Google's default encryption will be used.
+     */
+    public abstract Builder setEncryptionConfig(CustomerManagedEncryption encryptionConfig);
+
     abstract Builder setProto(com.google.spanner.admin.database.v1.Database proto);
 
     /** Builds the database from this builder. */
@@ -47,6 +57,7 @@ abstract static class BuilderImpl extends Builder {
     private RestoreInfo restoreInfo;
     private String versionRetentionPeriod;
     private Timestamp earliestVersionTime;
+    private CustomerManagedEncryption encryptionConfig;
     private com.google.spanner.admin.database.v1.Database proto;
 
     BuilderImpl(DatabaseId id) {
@@ -60,6 +71,7 @@ abstract static class BuilderImpl extends Builder {
       this.restoreInfo = other.restoreInfo;
       this.versionRetentionPeriod = other.versionRetentionPeriod;
       this.earliestVersionTime = other.earliestVersionTime;
+      this.encryptionConfig = other.encryptionConfig;
       this.proto = other.proto;
     }
 
@@ -93,6 +105,12 @@ Builder setEarliestVersionTime(Timestamp earliestVersionTime) {
       return this;
     }
 
+    @Override
+    public Builder setEncryptionConfig(@Nullable CustomerManagedEncryption encryptionConfig) {
+      this.encryptionConfig = encryptionConfig;
+      return this;
+    }
+
     @Override
     Builder setProto(@Nullable com.google.spanner.admin.database.v1.Database proto) {
       this.proto = proto;
@@ -118,6 +136,7 @@ public enum State {
   private final RestoreInfo restoreInfo;
   private final String versionRetentionPeriod;
   private final Timestamp earliestVersionTime;
+  private final CustomerManagedEncryption encryptionConfig;
   private final com.google.spanner.admin.database.v1.Database proto;
 
   public DatabaseInfo(DatabaseId id, State state) {
@@ -127,6 +146,7 @@ public DatabaseInfo(DatabaseId id, State state) {
     this.restoreInfo = null;
     this.versionRetentionPeriod = null;
     this.earliestVersionTime = null;
+    this.encryptionConfig = null;
     this.proto = null;
   }
 
@@ -137,6 +157,7 @@ public DatabaseInfo(DatabaseId id, State state) {
     this.restoreInfo = builder.restoreInfo;
     this.versionRetentionPeriod = builder.versionRetentionPeriod;
     this.earliestVersionTime = builder.earliestVersionTime;
+    this.encryptionConfig = builder.encryptionConfig;
     this.proto = builder.proto;
   }
 
@@ -180,6 +201,14 @@ public Timestamp getEarliestVersionTime() {
     return restoreInfo;
   }
 
+  /**
+   * Returns the {@link CustomerManagedEncryption} of the database if the database is encrypted, or
+   * null if this database is not encrypted.
+   */
+  public @Nullable CustomerManagedEncryption getEncryptionConfig() {
+    return encryptionConfig;
+  }
+
   /** Returns the raw proto instance that was used to construct this {@link Database}. */
   public @Nullable com.google.spanner.admin.database.v1.Database getProto() {
     return proto;
@@ -199,19 +228,32 @@ public boolean equals(Object o) {
         && Objects.equals(createTime, that.createTime)
         && Objects.equals(restoreInfo, that.restoreInfo)
         && Objects.equals(versionRetentionPeriod, that.versionRetentionPeriod)
-        && Objects.equals(earliestVersionTime, that.earliestVersionTime);
+        && Objects.equals(earliestVersionTime, that.earliestVersionTime)
+        && Objects.equals(encryptionConfig, that.encryptionConfig);
   }
 
   @Override
   public int hashCode() {
     return Objects.hash(
-        id, state, createTime, restoreInfo, versionRetentionPeriod, earliestVersionTime);
+        id,
+        state,
+        createTime,
+        restoreInfo,
+        versionRetentionPeriod,
+        earliestVersionTime,
+        encryptionConfig);
   }
 
   @Override
   public String toString() {
     return String.format(
-        "Database[%s, %s, %s, %s, %s, %s]",
-        id.getName(), state, createTime, restoreInfo, versionRetentionPeriod, earliestVersionTime);
+        "Database[%s, %s, %s, %s, %s, %s, %s]",
+        id.getName(),
+        state,
+        createTime,
+        restoreInfo,
+        versionRetentionPeriod,
+        earliestVersionTime,
+        encryptionConfig);
   }
 }
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Restore.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Restore.java
new file mode 100644
index 0000000000..d6a9c28850
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Restore.java
@@ -0,0 +1,109 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner;
+
+import com.google.cloud.spanner.encryption.RestoreEncryptionConfig;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.Objects;
+
+/** Represents a restore operation of a Cloud Spanner backup. */
+public class Restore {
+
+  public static class Builder {
+
+    private final BackupId source;
+    private final DatabaseId destination;
+    private RestoreEncryptionConfig encryptionConfig;
+
+    public Builder(BackupId source, DatabaseId destination) {
+      this.source = source;
+      this.destination = destination;
+    }
+
+    /**
+     * Optional for restoring a backup.
+     *
+     * 
The encryption configuration to be used for the backup. The possible configurations are
+     * {@link com.google.cloud.spanner.encryption.CustomerManagedEncryption}, {@link
+     * com.google.cloud.spanner.encryption.GoogleDefaultEncryption} and {@link
+     * com.google.cloud.spanner.encryption.UseBackupEncryption}.
+     *
+     * 
If no encryption config is given the database will be restored with the same encryption as
+     * set by the backup ({@link com.google.cloud.spanner.encryption.UseBackupEncryption}).
+     */
+    public Builder setEncryptionConfig(RestoreEncryptionConfig encryptionConfig) {
+      this.encryptionConfig = encryptionConfig;
+      return this;
+    }
+
+    public Restore build() {
+      return new Restore(this);
+    }
+  }
+
+  private final BackupId source;
+  private final DatabaseId destination;
+  private final RestoreEncryptionConfig encryptionConfig;
+
+  Restore(Builder builder) {
+    this(builder.source, builder.destination, builder.encryptionConfig);
+  }
+
+  @VisibleForTesting
+  Restore(BackupId source, DatabaseId destination, RestoreEncryptionConfig encryptionConfig) {
+    this.source = source;
+    this.destination = destination;
+    this.encryptionConfig = encryptionConfig;
+  }
+
+  public BackupId getSource() {
+    return source;
+  }
+
+  public DatabaseId getDestination() {
+    return destination;
+  }
+
+  public RestoreEncryptionConfig getEncryptionConfig() {
+    return encryptionConfig;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    Restore restore = (Restore) o;
+    return Objects.equals(source, restore.source)
+        && Objects.equals(destination, restore.destination)
+        && Objects.equals(encryptionConfig, restore.encryptionConfig);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(source, destination, encryptionConfig);
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "Restore[%s, %s, %s]", source.getName(), destination.getName(), encryptionConfig);
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/BackupEncryptionConfig.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/BackupEncryptionConfig.java
new file mode 100644
index 0000000000..a6e9fc1356
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/BackupEncryptionConfig.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+import com.google.api.core.InternalApi;
+
+/** Marker interface for encryption configurations that can be applied on backups. */
+@InternalApi
+public interface BackupEncryptionConfig {}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/CustomerManagedEncryption.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/CustomerManagedEncryption.java
new file mode 100644
index 0000000000..fdc48cf3d2
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/CustomerManagedEncryption.java
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+import com.google.spanner.admin.database.v1.EncryptionConfig;
+import java.util.Objects;
+
+/** The data is encrypted with a key provided by the customer. */
+public class CustomerManagedEncryption implements BackupEncryptionConfig, RestoreEncryptionConfig {
+
+  private final String kmsKeyName;
+
+  CustomerManagedEncryption(String kmsKeyName) {
+    this.kmsKeyName = kmsKeyName;
+  }
+
+  public String getKmsKeyName() {
+    return kmsKeyName;
+  }
+
+  /**
+   * Returns a {@link CustomerManagedEncryption} instance from the given proto, or null
+   * if the given proto is the default proto instance (i.e. there is no encryption config).
+   */
+  public static CustomerManagedEncryption fromProtoOrNull(EncryptionConfig proto) {
+    return proto.equals(EncryptionConfig.getDefaultInstance())
+        ? null
+        : new CustomerManagedEncryption(proto.getKmsKeyName());
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    CustomerManagedEncryption that = (CustomerManagedEncryption) o;
+    return Objects.equals(kmsKeyName, that.kmsKeyName);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(kmsKeyName);
+  }
+
+  @Override
+  public String toString() {
+    return "CustomerManagedEncryption{" + "kmsKeyName='" + kmsKeyName + '\'' + '}';
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapper.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapper.java
new file mode 100644
index 0000000000..0a18e9844a
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapper.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+import com.google.spanner.admin.database.v1.CreateBackupEncryptionConfig;
+import com.google.spanner.admin.database.v1.EncryptionConfig;
+import com.google.spanner.admin.database.v1.RestoreDatabaseEncryptionConfig;
+
+/** Maps encryption config domain classes to their protobuf counterpart. */
+public class EncryptionConfigProtoMapper {
+
+  /** Returns an encryption config to be used for a database. */
+  public static EncryptionConfig encryptionConfig(CustomerManagedEncryption config) {
+    return EncryptionConfig.newBuilder().setKmsKeyName(config.getKmsKeyName()).build();
+  }
+
+  /** Returns an encryption config to be used for a backup. */
+  public static CreateBackupEncryptionConfig createBackupEncryptionConfig(
+      BackupEncryptionConfig config) {
+    if (config instanceof CustomerManagedEncryption) {
+      return CreateBackupEncryptionConfig.newBuilder()
+          .setEncryptionType(
+              CreateBackupEncryptionConfig.EncryptionType.CUSTOMER_MANAGED_ENCRYPTION)
+          .setKmsKeyName(((CustomerManagedEncryption) config).getKmsKeyName())
+          .build();
+    } else if (config instanceof GoogleDefaultEncryption) {
+      return CreateBackupEncryptionConfig.newBuilder()
+          .setEncryptionType(CreateBackupEncryptionConfig.EncryptionType.GOOGLE_DEFAULT_ENCRYPTION)
+          .build();
+    } else if (config instanceof UseDatabaseEncryption) {
+      return CreateBackupEncryptionConfig.newBuilder()
+          .setEncryptionType(CreateBackupEncryptionConfig.EncryptionType.USE_DATABASE_ENCRYPTION)
+          .build();
+    } else {
+      throw new IllegalArgumentException("Unknown backup encryption configuration " + config);
+    }
+  }
+
+  /** Returns an encryption config to be used for a database restore. */
+  public static RestoreDatabaseEncryptionConfig restoreDatabaseEncryptionConfig(
+      RestoreEncryptionConfig config) {
+    if (config instanceof CustomerManagedEncryption) {
+      return RestoreDatabaseEncryptionConfig.newBuilder()
+          .setEncryptionType(
+              RestoreDatabaseEncryptionConfig.EncryptionType.CUSTOMER_MANAGED_ENCRYPTION)
+          .setKmsKeyName(((CustomerManagedEncryption) config).getKmsKeyName())
+          .build();
+    } else if (config instanceof GoogleDefaultEncryption) {
+      return RestoreDatabaseEncryptionConfig.newBuilder()
+          .setEncryptionType(
+              RestoreDatabaseEncryptionConfig.EncryptionType.GOOGLE_DEFAULT_ENCRYPTION)
+          .build();
+    } else if (config instanceof UseBackupEncryption) {
+      return RestoreDatabaseEncryptionConfig.newBuilder()
+          .setEncryptionType(
+              RestoreDatabaseEncryptionConfig.EncryptionType
+                  .USE_CONFIG_DEFAULT_OR_BACKUP_ENCRYPTION)
+          .build();
+    } else {
+      throw new IllegalArgumentException("Unknown restore encryption configuration " + config);
+    }
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigs.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigs.java
new file mode 100644
index 0000000000..6f77da2c87
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionConfigs.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+import com.google.api.client.util.Preconditions;
+
+/** Encryption configuration factory. */
+public class EncryptionConfigs {
+
+  /** Returns a customer managed encryption configuration for the given key. */
+  public static CustomerManagedEncryption customerManagedEncryption(String kmsKeyName) {
+    Preconditions.checkArgument(
+        kmsKeyName != null, "Customer managed encryption key name must not be null");
+    return new CustomerManagedEncryption(kmsKeyName);
+  }
+
+  /** Returns google default encryption configuration. */
+  public static GoogleDefaultEncryption googleDefaultEncryption() {
+    return GoogleDefaultEncryption.INSTANCE;
+  }
+
+  /** Returns use database encryption configuration. */
+  public static UseDatabaseEncryption useDatabaseEncryption() {
+    return UseDatabaseEncryption.INSTANCE;
+  }
+
+  /** Returns use backup encryption configuration. */
+  public static UseBackupEncryption useBackupEncryption() {
+    return UseBackupEncryption.INSTANCE;
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionInfo.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionInfo.java
new file mode 100644
index 0000000000..f811cfc101
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/EncryptionInfo.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2020 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.rpc.Status;
+import java.util.Objects;
+
+/** Represents the encryption information for a Cloud Spanner backup. */
+public class EncryptionInfo {
+
+  private final String kmsKeyVersion;
+  private final com.google.spanner.admin.database.v1.EncryptionInfo.Type encryptionType;
+  private final Status encryptionStatus;
+
+  public EncryptionInfo(com.google.spanner.admin.database.v1.EncryptionInfo proto) {
+    this(proto.getKmsKeyVersion(), proto.getEncryptionType(), proto.getEncryptionStatus());
+  }
+
+  @VisibleForTesting
+  public EncryptionInfo(
+      String kmsKeyVersion,
+      com.google.spanner.admin.database.v1.EncryptionInfo.Type encryptionType,
+      Status encryptionStatus) {
+    this.kmsKeyVersion = kmsKeyVersion;
+    this.encryptionType = encryptionType;
+    this.encryptionStatus = encryptionStatus;
+  }
+
+  /**
+   * Returns a {@link EncryptionInfo} instance from the given proto, or null if the
+   * given proto is the default proto instance (i.e. there is no encryption info).
+   */
+  public static EncryptionInfo fromProtoOrNull(
+      com.google.spanner.admin.database.v1.EncryptionInfo proto) {
+    return proto.equals(com.google.spanner.admin.database.v1.EncryptionInfo.getDefaultInstance())
+        ? null
+        : new EncryptionInfo(proto);
+  }
+
+  public String getKmsKeyVersion() {
+    return kmsKeyVersion;
+  }
+
+  public com.google.spanner.admin.database.v1.EncryptionInfo.Type getEncryptionType() {
+    return encryptionType;
+  }
+
+  public Status getEncryptionStatus() {
+    return encryptionStatus;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    EncryptionInfo that = (EncryptionInfo) o;
+    return Objects.equals(kmsKeyVersion, that.kmsKeyVersion)
+        && encryptionType == that.encryptionType
+        && Objects.equals(encryptionStatus, that.encryptionStatus);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(kmsKeyVersion, encryptionType, encryptionStatus);
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "EncryptionInfo[kmsKeyVersion=%s,encryptionType=%s,encryptionStatus=%s]",
+        kmsKeyVersion, encryptionType, encryptionStatus);
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/GoogleDefaultEncryption.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/GoogleDefaultEncryption.java
new file mode 100644
index 0000000000..fa03da8bd5
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/GoogleDefaultEncryption.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+/** The data is encrypted with a key that is fully managed by Google. */
+public class GoogleDefaultEncryption implements BackupEncryptionConfig, RestoreEncryptionConfig {
+
+  static final GoogleDefaultEncryption INSTANCE = new GoogleDefaultEncryption();
+
+  private GoogleDefaultEncryption() {}
+
+  @Override
+  public String toString() {
+    return "GoogleDefaultEncryption{}";
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/RestoreEncryptionConfig.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/RestoreEncryptionConfig.java
new file mode 100644
index 0000000000..b23fbe69d0
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/RestoreEncryptionConfig.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+import com.google.api.core.InternalApi;
+
+/** Marker interface for encryption configurations that can be applied on restores. */
+@InternalApi
+public interface RestoreEncryptionConfig {}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseBackupEncryption.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseBackupEncryption.java
new file mode 100644
index 0000000000..b3604597ab
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseBackupEncryption.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+/** The data is encrypted with the same configuration as specified by the backup being restored. */
+public class UseBackupEncryption implements RestoreEncryptionConfig {
+
+  static final UseBackupEncryption INSTANCE = new UseBackupEncryption();
+
+  private UseBackupEncryption() {}
+
+  @Override
+  public String toString() {
+    return "UseBackupEncryption{}";
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseDatabaseEncryption.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseDatabaseEncryption.java
new file mode 100644
index 0000000000..1fc7233496
--- /dev/null
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/encryption/UseDatabaseEncryption.java
@@ -0,0 +1,33 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.spanner.encryption;
+
+/**
+ * The data is encrypted with the same configuration as specified by the source database for a
+ * backup.
+ */
+public class UseDatabaseEncryption implements BackupEncryptionConfig {
+
+  static final UseDatabaseEncryption INSTANCE = new UseDatabaseEncryption();
+
+  private UseDatabaseEncryption() {}
+
+  @Override
+  public String toString() {
+    return "UseDatabaseEncryption{}";
+  }
+}
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
index 608d9e23d5..94a7a121d6 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/GapicSpannerRpc.java
@@ -57,6 +57,7 @@
 import com.google.cloud.grpc.GrpcTransportOptions;
 import com.google.cloud.spanner.AdminRequestsPerMinuteExceededException;
 import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.Restore;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.SpannerOptions;
@@ -69,6 +70,7 @@
 import com.google.cloud.spanner.admin.instance.v1.stub.GrpcInstanceAdminStub;
 import com.google.cloud.spanner.admin.instance.v1.stub.InstanceAdminStub;
 import com.google.cloud.spanner.admin.instance.v1.stub.InstanceAdminStubSettings;
+import com.google.cloud.spanner.encryption.EncryptionConfigProtoMapper;
 import com.google.cloud.spanner.v1.stub.GrpcSpannerStub;
 import com.google.cloud.spanner.v1.stub.SpannerStub;
 import com.google.cloud.spanner.v1.stub.SpannerStubSettings;
@@ -971,17 +973,22 @@ public ListDatabasesResponse call() throws Exception {
   public OperationFuture createDatabase(
       final String instanceName,
       String createDatabaseStatement,
-      Iterable additionalStatements)
+      Iterable additionalStatements,
+      com.google.cloud.spanner.Database databaseInfo)
       throws SpannerException {
     final String databaseId =
         createDatabaseStatement.substring(
             "CREATE DATABASE `".length(), createDatabaseStatement.length() - 1);
-    CreateDatabaseRequest request =
+    CreateDatabaseRequest.Builder requestBuilder =
         CreateDatabaseRequest.newBuilder()
             .setParent(instanceName)
             .setCreateStatement(createDatabaseStatement)
-            .addAllExtraStatements(additionalStatements)
-            .build();
+            .addAllExtraStatements(additionalStatements);
+    if (databaseInfo.getEncryptionConfig() != null) {
+      requestBuilder.setEncryptionConfig(
+          EncryptionConfigProtoMapper.encryptionConfig(databaseInfo.getEncryptionConfig()));
+    }
+    final CreateDatabaseRequest request = requestBuilder.build();
 
     OperationFutureCallable callable =
         new OperationFutureCallable(
@@ -1145,15 +1152,31 @@ public List call() throws Exception {
 
   @Override
   public OperationFuture createBackup(
-      final String instanceName, final String backupId, final Backup backup)
-      throws SpannerException {
-    CreateBackupRequest request =
+      final com.google.cloud.spanner.Backup backupInfo) throws SpannerException {
+    final String instanceName = backupInfo.getInstanceId().getName();
+    final String databaseName = backupInfo.getDatabase().getName();
+    final String backupId = backupInfo.getId().getBackup();
+    final Backup.Builder backupBuilder =
+        com.google.spanner.admin.database.v1.Backup.newBuilder()
+            .setDatabase(databaseName)
+            .setExpireTime(backupInfo.getExpireTime().toProto());
+    if (backupInfo.getVersionTime() != null) {
+      backupBuilder.setVersionTime(backupInfo.getVersionTime().toProto());
+    }
+    final Backup backup = backupBuilder.build();
+
+    final CreateBackupRequest.Builder requestBuilder =
         CreateBackupRequest.newBuilder()
             .setParent(instanceName)
             .setBackupId(backupId)
-            .setBackup(backup)
-            .build();
-    OperationFutureCallable callable =
+            .setBackup(backup);
+    if (backupInfo.getEncryptionConfig() != null) {
+      requestBuilder.setEncryptionConfig(
+          EncryptionConfigProtoMapper.createBackupEncryptionConfig(
+              backupInfo.getEncryptionConfig()));
+    }
+    final CreateBackupRequest request = requestBuilder.build();
+    final OperationFutureCallable callable =
         new OperationFutureCallable(
             databaseAdminStub.createBackupOperationCallable(),
             request,
@@ -1197,48 +1220,54 @@ public Timestamp apply(Operation input) {
   }
 
   @Override
-  public OperationFuture restoreDatabase(
-      final String databaseInstanceName, final String databaseId, String backupName) {
-    RestoreDatabaseRequest request =
+  public OperationFuture restoreDatabase(final Restore restore) {
+    final String databaseInstanceName = restore.getDestination().getInstanceId().getName();
+    final String databaseId = restore.getDestination().getDatabase();
+    final RestoreDatabaseRequest.Builder requestBuilder =
         RestoreDatabaseRequest.newBuilder()
             .setParent(databaseInstanceName)
             .setDatabaseId(databaseId)
-            .setBackup(backupName)
-            .build();
+            .setBackup(restore.getSource().getName());
+    if (restore.getEncryptionConfig() != null) {
+      requestBuilder.setEncryptionConfig(
+          EncryptionConfigProtoMapper.restoreDatabaseEncryptionConfig(
+              restore.getEncryptionConfig()));
+    }
 
-    OperationFutureCallable callable =
-        new OperationFutureCallable(
-            databaseAdminStub.restoreDatabaseOperationCallable(),
-            request,
-            DatabaseAdminGrpc.getRestoreDatabaseMethod(),
-            databaseInstanceName,
-            new OperationsLister() {
-              @Override
-              public Paginated listOperations(String nextPageToken) {
-                return listDatabaseOperations(
-                    databaseInstanceName,
-                    0,
-                    String.format(
-                        "(metadata.@type:type.googleapis.com/%s) AND (metadata.name:%s)",
-                        RestoreDatabaseMetadata.getDescriptor().getFullName(),
-                        String.format("%s/databases/%s", databaseInstanceName, databaseId)),
-                    nextPageToken);
-              }
-            },
-            new Function() {
-              @Override
-              public Timestamp apply(Operation input) {
-                try {
-                  return input
-                      .getMetadata()
-                      .unpack(RestoreDatabaseMetadata.class)
-                      .getProgress()
-                      .getStartTime();
-                } catch (InvalidProtocolBufferException e) {
-                  return null;
-                }
-              }
-            });
+    final OperationFutureCallable
+        callable =
+            new OperationFutureCallable(
+                databaseAdminStub.restoreDatabaseOperationCallable(),
+                requestBuilder.build(),
+                DatabaseAdminGrpc.getRestoreDatabaseMethod(),
+                databaseInstanceName,
+                new OperationsLister() {
+                  @Override
+                  public Paginated listOperations(String nextPageToken) {
+                    return listDatabaseOperations(
+                        databaseInstanceName,
+                        0,
+                        String.format(
+                            "(metadata.@type:type.googleapis.com/%s) AND (metadata.name:%s)",
+                            RestoreDatabaseMetadata.getDescriptor().getFullName(),
+                            String.format("%s/databases/%s", databaseInstanceName, databaseId)),
+                        nextPageToken);
+                  }
+                },
+                new Function() {
+                  @Override
+                  public Timestamp apply(Operation input) {
+                    try {
+                      return input
+                          .getMetadata()
+                          .unpack(RestoreDatabaseMetadata.class)
+                          .getProgress()
+                          .getStartTime();
+                    } catch (InvalidProtocolBufferException e) {
+                      return null;
+                    }
+                  }
+                });
     return RetryHelper.runWithRetries(
         callable,
         databaseAdminStubSettings
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/SpannerRpc.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/SpannerRpc.java
index 6b42c0a754..d4a9650830 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/SpannerRpc.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/spi/v1/SpannerRpc.java
@@ -22,6 +22,7 @@
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.ServerStream;
 import com.google.cloud.ServiceRpc;
+import com.google.cloud.spanner.Restore;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.admin.database.v1.stub.DatabaseAdminStub;
 import com.google.cloud.spanner.admin.instance.v1.stub.InstanceAdminStub;
@@ -198,7 +199,10 @@ Paginated listDatabases(String instanceName, int pageSize, @Nullable S
       throws SpannerException;
 
   OperationFuture createDatabase(
-      String instanceName, String createDatabaseStatement, Iterable additionalStatements)
+      String instanceName,
+      String createDatabaseStatement,
+      Iterable additionalStatements,
+      com.google.cloud.spanner.Database database)
       throws SpannerException;
 
   OperationFuture updateDatabaseDdl(
@@ -216,26 +220,22 @@ Paginated listBackups(
       throws SpannerException;
 
   /**
-   * Creates a new backup from the source database specified in the {@link Backup} instance.
+   * Creates a new backup from the source database specified in the {@link
+   * com.google.cloud.spanner.Backup} instance.
    *
-   * @param instanceName the name of the instance where the backup should be created.
-   * @param backupId the id of the backup to create.
-   * @param backup the backup to create. The database and expireTime fields of the backup must be
-   *     filled.
+   * @param backupInfo the backup to create. The instance, database and expireTime fields of the
+   *     backup must be filled.
    * @return the operation that monitors the backup creation.
    */
   OperationFuture createBackup(
-      String instanceName, String backupId, Backup backup) throws SpannerException;
+      com.google.cloud.spanner.Backup backupInfo) throws SpannerException;
 
   /**
    * Restore a backup into the given database.
    *
-   * @param instanceName Fully qualified name of instance where to restore the database
-   * @param databaseId DatabaseId to restore into
-   * @param backupName Fully qualified name of backup to restore from
+   * @param restore a {@link Restore} instance with the backup source and destination database
    */
-  OperationFuture restoreDatabase(
-      String instanceName, String databaseId, String backupName);
+  OperationFuture restoreDatabase(Restore restore);
 
   /** Gets the backup with the specified name. */
   Backup getBackup(String backupName) throws SpannerException;
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/BackupTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/BackupTest.java
index 80b99f679b..f2b479b666 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/BackupTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/BackupTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.spanner;
 
 import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.fail;
@@ -30,6 +31,9 @@
 import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.Backup.Builder;
 import com.google.cloud.spanner.BackupInfo.State;
+import com.google.cloud.spanner.encryption.EncryptionInfo;
+import com.google.rpc.Code;
+import com.google.rpc.Status;
 import java.util.Arrays;
 import org.junit.Before;
 import org.junit.Test;
@@ -42,11 +46,20 @@
 
 @RunWith(JUnit4.class)
 public class BackupTest {
+
   private static final String NAME =
       "projects/test-project/instances/test-instance/backups/backup-1";
   private static final String DB = "projects/test-project/instances/test-instance/databases/db-1";
   private static final Timestamp EXP_TIME = Timestamp.ofTimeSecondsAndNanos(1000L, 1000);
   private static final Timestamp VERSION_TIME = Timestamp.ofTimeSecondsAndNanos(2000L, 2000);
+  public static final String KMS_KEY_VERSION = "key-version";
+  private static final com.google.spanner.admin.database.v1.EncryptionInfo ENCRYPTION_INFO =
+      com.google.spanner.admin.database.v1.EncryptionInfo.newBuilder()
+          .setEncryptionType(
+              com.google.spanner.admin.database.v1.EncryptionInfo.Type.CUSTOMER_MANAGED_ENCRYPTION)
+          .setEncryptionStatus(Status.newBuilder().setCode(Code.OK.getNumber()))
+          .setKmsKeyVersion(KMS_KEY_VERSION)
+          .build();
 
   @Mock DatabaseAdminClient dbClient;
 
@@ -100,37 +113,6 @@ public void create() {
     verify(dbClient).createBackup(backup);
   }
 
-  @Test
-  public void createWithoutSource() {
-    Timestamp expireTime = Timestamp.now();
-    Backup backup =
-        dbClient
-            .newBackupBuilder(BackupId.of("test-project", "dest-instance", "backup-id"))
-            .setExpireTime(expireTime)
-            .build();
-    try {
-      backup.create();
-      fail("Expected exception");
-    } catch (IllegalStateException e) {
-      assertNotNull(e.getMessage());
-    }
-  }
-
-  @Test
-  public void createWithoutExpireTime() {
-    Backup backup =
-        dbClient
-            .newBackupBuilder(BackupId.of("test-project", "instance-id", "backup-id"))
-            .setDatabase(DatabaseId.of("test-project", "instance-id", "src-database"))
-            .build();
-    try {
-      backup.create();
-      fail("Expected exception");
-    } catch (IllegalStateException e) {
-      assertNotNull(e.getMessage());
-    }
-  }
-
   @Test
   public void createWithoutVersionTimeShouldSucceed() {
     final Timestamp expireTime = Timestamp.now();
@@ -318,6 +300,17 @@ public void fromProto() {
     assertThat(backup.getState()).isEqualTo(BackupInfo.State.CREATING);
     assertThat(backup.getExpireTime()).isEqualTo(EXP_TIME);
     assertThat(backup.getVersionTime()).isEqualTo(VERSION_TIME);
+    assertThat(backup.getEncryptionInfo())
+        .isEqualTo(EncryptionInfo.fromProtoOrNull(ENCRYPTION_INFO));
+  }
+
+  @Test
+  public void testEqualsAndHashCode() {
+    final Backup backup1 = createBackup();
+    final Backup backup2 = createBackup();
+
+    assertEquals(backup1, backup2);
+    assertEquals(backup1.hashCode(), backup2.hashCode());
   }
 
   private Backup createBackup() {
@@ -329,6 +322,7 @@ private Backup createBackup() {
                 com.google.protobuf.Timestamp.newBuilder().setSeconds(1000L).setNanos(1000).build())
             .setVersionTime(
                 com.google.protobuf.Timestamp.newBuilder().setSeconds(2000L).setNanos(2000).build())
+            .setEncryptionInfo(ENCRYPTION_INFO)
             .setState(com.google.spanner.admin.database.v1.Backup.State.CREATING)
             .build();
     return Backup.fromProto(proto, dbClient);
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseAdminClientImplTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseAdminClientImplTest.java
index fb617797aa..b782847629 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseAdminClientImplTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseAdminClientImplTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.spanner;
 
 import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.mockito.MockitoAnnotations.initMocks;
@@ -25,6 +26,8 @@
 import com.google.cloud.Identity;
 import com.google.cloud.Role;
 import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.DatabaseInfo.State;
+import com.google.cloud.spanner.encryption.EncryptionConfigs;
 import com.google.cloud.spanner.spi.v1.SpannerRpc;
 import com.google.cloud.spanner.spi.v1.SpannerRpc.Paginated;
 import com.google.common.collect.ImmutableList;
@@ -43,6 +46,7 @@
 import com.google.spanner.admin.database.v1.CreateBackupMetadata;
 import com.google.spanner.admin.database.v1.CreateDatabaseMetadata;
 import com.google.spanner.admin.database.v1.Database;
+import com.google.spanner.admin.database.v1.EncryptionInfo;
 import com.google.spanner.admin.database.v1.RestoreDatabaseMetadata;
 import com.google.spanner.admin.database.v1.UpdateDatabaseDdlMetadata;
 import java.util.Arrays;
@@ -50,14 +54,12 @@
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.mockito.Mock;
 
-/** Unit tests for {@link com.google.cloud.spanner.SpannerImpl.DatabaseAdminClientImpl}. */
 @RunWith(JUnit4.class)
 public class DatabaseAdminClientImplTest {
   private static final String PROJECT_ID = "my-project";
@@ -72,6 +74,9 @@ public class DatabaseAdminClientImplTest {
   private static final String BK_NAME2 = "projects/my-project/instances/my-instance/backups/my-bk2";
   private static final Timestamp EARLIEST_VERSION_TIME = Timestamp.now();
   private static final String VERSION_RETENTION_PERIOD = "7d";
+  private static final String KMS_KEY_NAME =
+      "projects/my-project/locations/some-location/keyRings/my-keyring/cryptoKeys/my-key";
+  private static final String KMS_KEY_VERSION = "1";
 
   @Mock SpannerRpc rpc;
   DatabaseAdminClientImpl client;
@@ -91,6 +96,16 @@ private Database getDatabaseProto() {
         .build();
   }
 
+  private Database getEncryptedDatabaseProto() {
+    return getDatabaseProto()
+        .toBuilder()
+        .setEncryptionConfig(
+            com.google.spanner.admin.database.v1.EncryptionConfig.newBuilder()
+                .setKmsKeyName(KMS_KEY_NAME)
+                .build())
+        .build();
+  }
+
   private Database getAnotherDatabaseProto() {
     return Database.newBuilder().setName(DB_NAME2).setState(Database.State.READY).build();
   }
@@ -110,6 +125,15 @@ private Backup getBackupProto() {
         .build();
   }
 
+  private Backup getEncryptedBackupProto() {
+    return Backup.newBuilder()
+        .setName(BK_NAME)
+        .setDatabase(DB_NAME)
+        .setState(Backup.State.READY)
+        .setEncryptionInfo(EncryptionInfo.newBuilder().setKmsKeyVersion(KMS_KEY_VERSION).build())
+        .build();
+  }
+
   private Backup getAnotherBackupProto() {
     return Backup.newBuilder()
         .setName(BK_NAME2)
@@ -134,7 +158,11 @@ public void createDatabase() throws Exception {
         OperationFutureUtil.immediateOperationFuture(
             "createDatabase", getDatabaseProto(), CreateDatabaseMetadata.getDefaultInstance());
     when(rpc.createDatabase(
-            INSTANCE_NAME, "CREATE DATABASE `" + DB_ID + "`", Collections.emptyList()))
+            INSTANCE_NAME,
+            "CREATE DATABASE `" + DB_ID + "`",
+            Collections.emptyList(),
+            new com.google.cloud.spanner.Database(
+                DatabaseId.of(DB_NAME), State.UNSPECIFIED, client)))
         .thenReturn(rawOperationFuture);
     OperationFuture op =
         client.createDatabase(INSTANCE_ID, DB_ID, Collections.emptyList());
@@ -142,6 +170,31 @@ public void createDatabase() throws Exception {
     assertThat(op.get().getId().getName()).isEqualTo(DB_NAME);
   }
 
+  @Test
+  public void createEncryptedDatabase() throws Exception {
+    com.google.cloud.spanner.Database database =
+        client
+            .newDatabaseBuilder(DatabaseId.of(DB_NAME))
+            .setEncryptionConfig(EncryptionConfigs.customerManagedEncryption(KMS_KEY_NAME))
+            .build();
+
+    OperationFuture rawOperationFuture =
+        OperationFutureUtil.immediateOperationFuture(
+            "createDatabase",
+            getEncryptedDatabaseProto(),
+            CreateDatabaseMetadata.getDefaultInstance());
+    when(rpc.createDatabase(
+            INSTANCE_NAME,
+            "CREATE DATABASE `" + DB_ID + "`",
+            Collections.emptyList(),
+            database))
+        .thenReturn(rawOperationFuture);
+    OperationFuture op =
+        client.createDatabase(database, Collections.emptyList());
+    assertThat(op.isDone()).isTrue();
+    assertThat(op.get().getId().getName()).isEqualTo(DB_NAME);
+  }
+
   @Test
   public void updateDatabaseDdl() throws Exception {
     String opName = DB_NAME + "/operations/myop";
@@ -208,7 +261,7 @@ public void listDatabasesError() {
             SpannerExceptionFactory.newSpannerException(ErrorCode.INVALID_ARGUMENT, "Test error"));
     try {
       client.listDatabases(INSTANCE_ID, Options.pageSize(1));
-      Assert.fail("Missing expected exception");
+      fail("Missing expected exception");
     } catch (SpannerException e) {
       assertThat(e.getMessage()).contains(INSTANCE_NAME);
       // Assert that the call was done without a page token.
@@ -226,7 +279,7 @@ public void listDatabaseErrorWithToken() {
             SpannerExceptionFactory.newSpannerException(ErrorCode.INVALID_ARGUMENT, "Test error"));
     try {
       Lists.newArrayList(client.listDatabases(INSTANCE_ID, Options.pageSize(1)).iterateAll());
-      Assert.fail("Missing expected exception");
+      fail("Missing expected exception");
     } catch (SpannerException e) {
       assertThat(e.getMessage()).contains(INSTANCE_NAME);
       // Assert that the call was done without a page token.
@@ -310,8 +363,13 @@ public void createBackupWithParams() throws Exception {
         Timestamp.ofTimeMicroseconds(
             TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis())
                 + TimeUnit.HOURS.toMicros(28));
-    Backup backup = Backup.newBuilder().setDatabase(DB_NAME).setExpireTime(t.toProto()).build();
-    when(rpc.createBackup(INSTANCE_NAME, BK_ID, backup)).thenReturn(rawOperationFuture);
+    final com.google.cloud.spanner.Backup backup =
+        client
+            .newBackupBuilder(BackupId.of(PROJECT_ID, INSTANCE_ID, BK_ID))
+            .setDatabase(DatabaseId.of(PROJECT_ID, INSTANCE_ID, DB_ID))
+            .setExpireTime(t)
+            .build();
+    when(rpc.createBackup(backup)).thenReturn(rawOperationFuture);
     OperationFuture op =
         client.createBackup(INSTANCE_ID, BK_ID, DB_ID, t);
     assertThat(op.isDone()).isTrue();
@@ -330,12 +388,6 @@ public void createBackupWithBackupObject() throws ExecutionException, Interrupte
     final Timestamp versionTime =
         Timestamp.ofTimeMicroseconds(
             TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis()) - TimeUnit.DAYS.toMicros(2));
-    final Backup expectedCallBackup =
-        Backup.newBuilder()
-            .setDatabase(DB_NAME)
-            .setExpireTime(expireTime.toProto())
-            .setVersionTime(versionTime.toProto())
-            .build();
     final com.google.cloud.spanner.Backup requestBackup =
         client
             .newBackupBuilder(BackupId.of(PROJECT_ID, INSTANCE_ID, BK_ID))
@@ -344,7 +396,7 @@ public void createBackupWithBackupObject() throws ExecutionException, Interrupte
             .setVersionTime(versionTime)
             .build();
 
-    when(rpc.createBackup(INSTANCE_NAME, BK_ID, expectedCallBackup)).thenReturn(rawOperationFuture);
+    when(rpc.createBackup(requestBackup)).thenReturn(rawOperationFuture);
 
     final OperationFuture op =
         client.createBackup(requestBackup);
@@ -352,6 +404,52 @@ public void createBackupWithBackupObject() throws ExecutionException, Interrupte
     assertThat(op.get().getId().getName()).isEqualTo(BK_NAME);
   }
 
+  @Test(expected = IllegalArgumentException.class)
+  public void testCreateBackupNoExpireTime() {
+    final com.google.cloud.spanner.Backup requestBackup =
+        client
+            .newBackupBuilder(BackupId.of(PROJECT_ID, INSTANCE_ID, BK_ID))
+            .setDatabase(DatabaseId.of(PROJECT_ID, INSTANCE_ID, DB_ID))
+            .build();
+
+    client.createBackup(requestBackup);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testCreateBackupNoDatabase() {
+    final com.google.cloud.spanner.Backup requestBackup =
+        client
+            .newBackupBuilder(BackupId.of(PROJECT_ID, INSTANCE_ID, BK_ID))
+            .setExpireTime(Timestamp.now())
+            .build();
+
+    client.createBackup(requestBackup);
+  }
+
+  @Test
+  public void createEncryptedBackup() throws ExecutionException, InterruptedException {
+    final OperationFuture rawOperationFuture =
+        OperationFutureUtil.immediateOperationFuture(
+            "createBackup", getEncryptedBackupProto(), CreateBackupMetadata.getDefaultInstance());
+    final Timestamp t =
+        Timestamp.ofTimeMicroseconds(
+            TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis())
+                + TimeUnit.HOURS.toMicros(28));
+    final com.google.cloud.spanner.Backup backup =
+        client
+            .newBackupBuilder(BackupId.of(PROJECT_ID, INSTANCE_ID, BK_ID))
+            .setDatabase(DatabaseId.of(PROJECT_ID, INSTANCE_ID, DB_ID))
+            .setExpireTime(t)
+            .setEncryptionConfig(EncryptionConfigs.customerManagedEncryption(KMS_KEY_NAME))
+            .build();
+    when(rpc.createBackup(backup)).thenReturn(rawOperationFuture);
+    final OperationFuture op =
+        client.createBackup(backup);
+    assertThat(op.isDone()).isTrue();
+    assertThat(op.get().getId().getName()).isEqualTo(BK_NAME);
+    assertThat(op.get().getEncryptionInfo().getKmsKeyVersion()).isEqualTo(KMS_KEY_VERSION);
+  }
+
   @Test
   public void deleteBackup() {
     client.deleteBackup(INSTANCE_ID, BK_ID);
@@ -404,10 +502,35 @@ public void restoreDatabase() throws Exception {
     OperationFuture rawOperationFuture =
         OperationFutureUtil.immediateOperationFuture(
             "restoreDatabase", getDatabaseProto(), RestoreDatabaseMetadata.getDefaultInstance());
-    when(rpc.restoreDatabase(INSTANCE_NAME, DB_ID, BK_NAME)).thenReturn(rawOperationFuture);
+    final Restore restore =
+        new Restore.Builder(
+                BackupId.of(PROJECT_ID, INSTANCE_ID, BK_ID),
+                DatabaseId.of(PROJECT_ID, INSTANCE_ID, DB_ID))
+            .build();
+    when(rpc.restoreDatabase(restore)).thenReturn(rawOperationFuture);
+    OperationFuture op =
+        client.restoreDatabase(restore);
+    assertThat(op.isDone()).isTrue();
+    assertThat(op.get().getId().getName()).isEqualTo(DB_NAME);
+  }
+
+  @Test
+  public void restoreEncryptedDatabase() throws Exception {
+    OperationFuture rawOperationFuture =
+        OperationFutureUtil.immediateOperationFuture(
+            "restoreEncryptedDatabase",
+            getEncryptedDatabaseProto(),
+            RestoreDatabaseMetadata.getDefaultInstance());
+    final Restore restore =
+        new Restore.Builder(
+                BackupId.of(PROJECT_ID, INSTANCE_ID, BK_ID),
+                DatabaseId.of(PROJECT_ID, INSTANCE_ID, DB_ID))
+            .build();
+    when(rpc.restoreDatabase(restore)).thenReturn(rawOperationFuture);
     OperationFuture op =
-        client.restoreDatabase(INSTANCE_ID, BK_ID, INSTANCE_ID, DB_ID);
+        client.restoreDatabase(restore);
     assertThat(op.isDone()).isTrue();
     assertThat(op.get().getId().getName()).isEqualTo(DB_NAME);
+    assertThat(op.get().getEncryptionConfig().getKmsKeyName()).isEqualTo(KMS_KEY_NAME);
   }
 }
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseTest.java
index 7c61720b21..7b3b533874 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseTest.java
@@ -17,6 +17,7 @@
 package com.google.cloud.spanner;
 
 import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.mockito.MockitoAnnotations.initMocks;
@@ -26,7 +27,13 @@
 import com.google.cloud.Role;
 import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.DatabaseInfo.State;
+import com.google.cloud.spanner.encryption.EncryptionConfigs;
+import com.google.rpc.Code;
+import com.google.rpc.Status;
+import com.google.spanner.admin.database.v1.EncryptionInfo;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -44,6 +51,19 @@ public class DatabaseTest {
 
   private static final Timestamp EARLIEST_VERSION_TIME = Timestamp.now();
   private static final String VERSION_RETENTION_PERIOD = "7d";
+  private static final String KMS_KEY_NAME = "kms-key-name";
+  private static final String KMS_KEY_VERSION = "kms-key-version";
+  private static final com.google.spanner.admin.database.v1.EncryptionConfig ENCRYPTION_CONFIG =
+      com.google.spanner.admin.database.v1.EncryptionConfig.newBuilder()
+          .setKmsKeyName(KMS_KEY_NAME)
+          .build();
+  private static final List ENCRYPTION_INFOS =
+      Collections.singletonList(
+          EncryptionInfo.newBuilder()
+              .setEncryptionType(EncryptionInfo.Type.CUSTOMER_MANAGED_ENCRYPTION)
+              .setEncryptionStatus(Status.newBuilder().setCode(Code.OK.getNumber()))
+              .setKmsKeyVersion(KMS_KEY_VERSION)
+              .build());
 
   @Mock DatabaseAdminClient dbClient;
 
@@ -58,6 +78,14 @@ public Backup.Builder answer(InvocationOnMock invocation) {
                 return new Backup.Builder(dbClient, (BackupId) invocation.getArguments()[0]);
               }
             });
+    when(dbClient.newDatabaseBuilder(Mockito.any(DatabaseId.class)))
+        .thenAnswer(
+            new Answer() {
+              @Override
+              public Database.Builder answer(InvocationOnMock invocation) throws Throwable {
+                return new Database.Builder(dbClient, (DatabaseId) invocation.getArguments()[0]);
+              }
+            });
   }
 
   @Test
@@ -88,6 +116,38 @@ public void fromProto() {
     assertThat(db.getState()).isEqualTo(DatabaseInfo.State.CREATING);
     assertThat(db.getVersionRetentionPeriod()).isEqualTo(VERSION_RETENTION_PERIOD);
     assertThat(db.getEarliestVersionTime()).isEqualTo(EARLIEST_VERSION_TIME);
+    assertThat(db.getEncryptionConfig())
+        .isEqualTo(EncryptionConfigs.customerManagedEncryption(KMS_KEY_NAME));
+  }
+
+  @Test
+  public void testFromProtoWithEncryptionConfig() {
+    com.google.spanner.admin.database.v1.Database proto =
+        com.google.spanner.admin.database.v1.Database.newBuilder()
+            .setName(NAME)
+            .setEncryptionConfig(
+                com.google.spanner.admin.database.v1.EncryptionConfig.newBuilder()
+                    .setKmsKeyName("some-key")
+                    .build())
+            .build();
+    Database db = Database.fromProto(proto, dbClient);
+    assertThat(db.getEncryptionConfig()).isNotNull();
+    assertThat(db.getEncryptionConfig().getKmsKeyName()).isEqualTo("some-key");
+  }
+
+  @Test
+  public void testBuildWithEncryptionConfig() {
+    Database db =
+        dbClient
+            .newDatabaseBuilder(DatabaseId.of("my-project", "my-instance", "my-database"))
+            .setEncryptionConfig(
+                EncryptionConfigs.customerManagedEncryption(
+                    "projects/my-project/locations/some-location/keyRings/my-keyring/cryptoKeys/my-key"))
+            .build();
+    assertThat(db.getEncryptionConfig()).isNotNull();
+    assertThat(db.getEncryptionConfig().getKmsKeyName())
+        .isEqualTo(
+            "projects/my-project/locations/some-location/keyRings/my-keyring/cryptoKeys/my-key");
   }
 
   @Test
@@ -120,6 +180,15 @@ public void testIAMPermissions() {
     verify(dbClient).testDatabaseIAMPermissions("test-instance", "test-database", permissions);
   }
 
+  @Test
+  public void testEqualsAndHashCode() {
+    final Database database1 = createDatabase();
+    final Database database2 = createDatabase();
+
+    assertEquals(database1, database2);
+    assertEquals(database1.hashCode(), database2.hashCode());
+  }
+
   private Database createDatabase() {
     com.google.spanner.admin.database.v1.Database proto =
         com.google.spanner.admin.database.v1.Database.newBuilder()
@@ -127,6 +196,8 @@ private Database createDatabase() {
             .setState(com.google.spanner.admin.database.v1.Database.State.CREATING)
             .setEarliestVersionTime(EARLIEST_VERSION_TIME.toProto())
             .setVersionRetentionPeriod(VERSION_RETENTION_PERIOD)
+            .setEncryptionConfig(ENCRYPTION_CONFIG)
+            .addAllEncryptionInfo(ENCRYPTION_INFOS)
             .build();
     return Database.fromProto(proto, dbClient);
   }
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/RestoreTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/RestoreTest.java
new file mode 100644
index 0000000000..409efb2046
--- /dev/null
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/RestoreTest.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.spanner;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.cloud.spanner.encryption.EncryptionConfigs;
+import com.google.cloud.spanner.encryption.RestoreEncryptionConfig;
+import org.junit.Test;
+
+/** Unit tests for {@link com.google.cloud.spanner.Restore} */
+public class RestoreTest {
+
+  private static final BackupId BACKUP_ID =
+      BackupId.of("test-project", "test-instance", "test-backup");
+  private static final DatabaseId DATABASE_ID =
+      DatabaseId.of("test-project", "test-instance", "test-database");
+  private static final String KMS_KEY_NAME = "kms-key-name";
+  private static final RestoreEncryptionConfig ENCRYPTION_CONFIG_INFO =
+      EncryptionConfigs.customerManagedEncryption(KMS_KEY_NAME);
+
+  @Test
+  public void testRestore() {
+    final Restore actualRestore =
+        new Restore.Builder(BACKUP_ID, DATABASE_ID)
+            .setEncryptionConfig(ENCRYPTION_CONFIG_INFO)
+            .build();
+    final Restore expectedRestore = new Restore(BACKUP_ID, DATABASE_ID, ENCRYPTION_CONFIG_INFO);
+
+    assertEquals(expectedRestore, actualRestore);
+  }
+
+  @Test
+  public void testEqualsAndHashCode() {
+    final Restore restore1 = new Restore(BACKUP_ID, DATABASE_ID, ENCRYPTION_CONFIG_INFO);
+    final Restore restore2 = new Restore(BACKUP_ID, DATABASE_ID, ENCRYPTION_CONFIG_INFO);
+
+    assertEquals(restore1, restore2);
+    assertEquals(restore1.hashCode(), restore2.hashCode());
+  }
+}
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/CustomerManagedEncryptionTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/CustomerManagedEncryptionTest.java
new file mode 100644
index 0000000000..5a8fe204fb
--- /dev/null
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/CustomerManagedEncryptionTest.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.spanner.encryption;
+
+import static org.junit.Assert.*;
+
+import com.google.spanner.admin.database.v1.EncryptionConfig;
+import org.junit.Test;
+
+/** Unit tests for {@link CustomerManagedEncryption}. */
+public class CustomerManagedEncryptionTest {
+
+  @Test
+  public void testFromProtoWithDefaultInstance() {
+    final CustomerManagedEncryption actual =
+        CustomerManagedEncryption.fromProtoOrNull(EncryptionConfig.getDefaultInstance());
+
+    assertNull(actual);
+  }
+
+  @Test
+  public void testFromProto() {
+    final CustomerManagedEncryption expected = new CustomerManagedEncryption("kms-key-name");
+    final EncryptionConfig encryptionConfig =
+        EncryptionConfig.newBuilder().setKmsKeyName("kms-key-name").build();
+
+    final CustomerManagedEncryption actual =
+        CustomerManagedEncryption.fromProtoOrNull(encryptionConfig);
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void testEqualsAndHashCode() {
+    final CustomerManagedEncryption customerManagedEncryption1 =
+        new CustomerManagedEncryption("kms-key-name");
+    final CustomerManagedEncryption customerManagedEncryption2 =
+        new CustomerManagedEncryption("kms-key-name");
+
+    assertEquals(customerManagedEncryption1, customerManagedEncryption2);
+    assertEquals(customerManagedEncryption1.hashCode(), customerManagedEncryption2.hashCode());
+  }
+}
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapperTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapperTest.java
new file mode 100644
index 0000000000..4ce300c2f6
--- /dev/null
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigProtoMapperTest.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.spanner.encryption;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.spanner.admin.database.v1.CreateBackupEncryptionConfig;
+import com.google.spanner.admin.database.v1.EncryptionConfig;
+import com.google.spanner.admin.database.v1.RestoreDatabaseEncryptionConfig;
+import org.junit.Test;
+
+/** Unit tests for {@link com.google.cloud.spanner.encryption.EncryptionConfigProtoMapper} */
+public class EncryptionConfigProtoMapperTest {
+
+  public static final String KMS_KEY_NAME = "kms-key-name";
+
+  @Test
+  public void testEncryptionConfig() {
+    final EncryptionConfig expected =
+        EncryptionConfig.newBuilder().setKmsKeyName(KMS_KEY_NAME).build();
+
+    final EncryptionConfig actual =
+        EncryptionConfigProtoMapper.encryptionConfig(new CustomerManagedEncryption(KMS_KEY_NAME));
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void testCreateBackupConfigCustomerManagedEncryption() {
+    final CreateBackupEncryptionConfig expected =
+        CreateBackupEncryptionConfig.newBuilder()
+            .setEncryptionType(
+                CreateBackupEncryptionConfig.EncryptionType.CUSTOMER_MANAGED_ENCRYPTION)
+            .setKmsKeyName(KMS_KEY_NAME)
+            .build();
+
+    final CreateBackupEncryptionConfig actual =
+        EncryptionConfigProtoMapper.createBackupEncryptionConfig(
+            new CustomerManagedEncryption(KMS_KEY_NAME));
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void testCreateBackupConfigGoogleDefaultEncryption() {
+    final CreateBackupEncryptionConfig expected =
+        CreateBackupEncryptionConfig.newBuilder()
+            .setEncryptionType(
+                CreateBackupEncryptionConfig.EncryptionType.GOOGLE_DEFAULT_ENCRYPTION)
+            .build();
+
+    final CreateBackupEncryptionConfig actual =
+        EncryptionConfigProtoMapper.createBackupEncryptionConfig(GoogleDefaultEncryption.INSTANCE);
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void testCreateBackupConfigUseDatabaseEncryption() {
+    final CreateBackupEncryptionConfig expected =
+        CreateBackupEncryptionConfig.newBuilder()
+            .setEncryptionType(CreateBackupEncryptionConfig.EncryptionType.USE_DATABASE_ENCRYPTION)
+            .build();
+
+    final CreateBackupEncryptionConfig actual =
+        EncryptionConfigProtoMapper.createBackupEncryptionConfig(UseDatabaseEncryption.INSTANCE);
+
+    assertEquals(expected, actual);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testCreateBackupInvalidEncryption() {
+    EncryptionConfigProtoMapper.createBackupEncryptionConfig(null);
+  }
+
+  @Test
+  public void testRestoreDatabaseConfigCustomerManagedEncryption() {
+    final RestoreDatabaseEncryptionConfig expected =
+        RestoreDatabaseEncryptionConfig.newBuilder()
+            .setEncryptionType(
+                RestoreDatabaseEncryptionConfig.EncryptionType.CUSTOMER_MANAGED_ENCRYPTION)
+            .setKmsKeyName(KMS_KEY_NAME)
+            .build();
+
+    final RestoreDatabaseEncryptionConfig actual =
+        EncryptionConfigProtoMapper.restoreDatabaseEncryptionConfig(
+            new CustomerManagedEncryption(KMS_KEY_NAME));
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void testRestoreDatabaseConfigGoogleDefaultEncryption() {
+    final RestoreDatabaseEncryptionConfig expected =
+        RestoreDatabaseEncryptionConfig.newBuilder()
+            .setEncryptionType(
+                RestoreDatabaseEncryptionConfig.EncryptionType.GOOGLE_DEFAULT_ENCRYPTION)
+            .build();
+
+    final RestoreDatabaseEncryptionConfig actual =
+        EncryptionConfigProtoMapper.restoreDatabaseEncryptionConfig(
+            GoogleDefaultEncryption.INSTANCE);
+
+    assertEquals(expected, actual);
+  }
+
+  @Test
+  public void testRestoreDatabaseConfigUseBackupEncryption() {
+    final RestoreDatabaseEncryptionConfig expected =
+        RestoreDatabaseEncryptionConfig.newBuilder()
+            .setEncryptionType(
+                RestoreDatabaseEncryptionConfig.EncryptionType
+                    .USE_CONFIG_DEFAULT_OR_BACKUP_ENCRYPTION)
+            .build();
+
+    final RestoreDatabaseEncryptionConfig actual =
+        EncryptionConfigProtoMapper.restoreDatabaseEncryptionConfig(UseBackupEncryption.INSTANCE);
+
+    assertEquals(expected, actual);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testRestoreDatabaseConfigInvalidEncryption() {
+    EncryptionConfigProtoMapper.restoreDatabaseEncryptionConfig(null);
+  }
+}
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigsTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigsTest.java
new file mode 100644
index 0000000000..82f997a1c4
--- /dev/null
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionConfigsTest.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.spanner.encryption;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+import org.junit.Test;
+
+/** Unit tests for {@link EncryptionConfigs} */
+public class EncryptionConfigsTest {
+
+  @Test
+  public void testCustomerManagedEncryption() {
+    final CustomerManagedEncryption expected = new CustomerManagedEncryption("kms-key-name");
+
+    final CustomerManagedEncryption actual =
+        EncryptionConfigs.customerManagedEncryption("kms-key-name");
+
+    assertEquals(expected, actual);
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testCustomerManagedEncryptionNullKeyName() {
+    EncryptionConfigs.customerManagedEncryption(null);
+  }
+
+  @Test
+  public void testGoogleDefaultEncryption() {
+    assertSame(EncryptionConfigs.googleDefaultEncryption(), GoogleDefaultEncryption.INSTANCE);
+  }
+
+  @Test
+  public void testUseDatabaseEncryption() {
+    assertSame(EncryptionConfigs.useDatabaseEncryption(), UseDatabaseEncryption.INSTANCE);
+  }
+
+  @Test
+  public void testUseBackupEncryption() {
+    assertSame(EncryptionConfigs.useBackupEncryption(), UseBackupEncryption.INSTANCE);
+  }
+}
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionInfoTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionInfoTest.java
new file mode 100644
index 0000000000..88a11d19c8
--- /dev/null
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/encryption/EncryptionInfoTest.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.spanner.encryption;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+import com.google.rpc.Code;
+import com.google.rpc.Status;
+import org.junit.Test;
+
+/** Unit tests for {@link com.google.cloud.spanner.encryption.EncryptionInfo} */
+public class EncryptionInfoTest {
+
+  private static final String KMS_KEY_VERSION = "kms-key-version";
+  private static final com.google.spanner.admin.database.v1.EncryptionInfo.Type
+      CUSTOMER_MANAGED_ENCRYPTION =
+          com.google.spanner.admin.database.v1.EncryptionInfo.Type.CUSTOMER_MANAGED_ENCRYPTION;
+  private static final Status OK_STATUS = Status.newBuilder().setCode(Code.OK_VALUE).build();
+
+  @Test
+  public void testEncryptionInfoFromProtoDefaultInstance() {
+    final EncryptionInfo encryptionInfo =
+        EncryptionInfo.fromProtoOrNull(
+            com.google.spanner.admin.database.v1.EncryptionInfo.getDefaultInstance());
+
+    assertNull(encryptionInfo);
+  }
+
+  @Test
+  public void testEncryptionInfoFromProto() {
+    final EncryptionInfo actualEncryptionInfo =
+        EncryptionInfo.fromProtoOrNull(
+            com.google.spanner.admin.database.v1.EncryptionInfo.newBuilder()
+                .setEncryptionStatus(OK_STATUS)
+                .setEncryptionTypeValue(CUSTOMER_MANAGED_ENCRYPTION.getNumber())
+                .setKmsKeyVersion(KMS_KEY_VERSION)
+                .build());
+
+    final EncryptionInfo expectedEncryptionInfo =
+        new EncryptionInfo(KMS_KEY_VERSION, CUSTOMER_MANAGED_ENCRYPTION, OK_STATUS);
+
+    assertEquals(expectedEncryptionInfo, actualEncryptionInfo);
+  }
+
+  @Test
+  public void testEqualsAndHashCode() {
+    final EncryptionInfo encryptionInfo1 =
+        EncryptionInfo.fromProtoOrNull(
+            com.google.spanner.admin.database.v1.EncryptionInfo.newBuilder()
+                .setEncryptionStatus(OK_STATUS)
+                .setEncryptionTypeValue(CUSTOMER_MANAGED_ENCRYPTION.getNumber())
+                .setKmsKeyVersion(KMS_KEY_VERSION)
+                .build());
+    final EncryptionInfo encryptionInfo2 =
+        EncryptionInfo.fromProtoOrNull(
+            com.google.spanner.admin.database.v1.EncryptionInfo.newBuilder()
+                .setEncryptionStatus(OK_STATUS)
+                .setEncryptionTypeValue(CUSTOMER_MANAGED_ENCRYPTION.getNumber())
+                .setKmsKeyVersion(KMS_KEY_VERSION)
+                .build());
+
+    assertEquals(encryptionInfo1, encryptionInfo2);
+    assertEquals(encryptionInfo1.hashCode(), encryptionInfo2.hashCode());
+  }
+}
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/it/ITBackupTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/it/ITBackupTest.java
index 786e7b3d6d..53dcc7907f 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/it/ITBackupTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/it/ITBackupTest.java
@@ -42,11 +42,14 @@
 import com.google.cloud.spanner.Mutation;
 import com.google.cloud.spanner.Options;
 import com.google.cloud.spanner.ParallelIntegrationTest;
+import com.google.cloud.spanner.Restore;
 import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.encryption.EncryptionConfigs;
 import com.google.cloud.spanner.testing.RemoteSpannerHelper;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.Iterables;
@@ -59,6 +62,7 @@
 import io.grpc.Status;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.ExecutionException;
@@ -86,7 +90,9 @@
 public class ITBackupTest {
   private static final Logger logger = Logger.getLogger(ITBackupTest.class.getName());
   private static final String EXPECTED_OP_NAME_FORMAT = "%s/backups/%s/operations/";
+  private static final String KMS_KEY_NAME_PROPERTY = "spanner.testenv.kms_key.name";
   @ClassRule public static IntegrationTestEnv env = new IntegrationTestEnv();
+  private static String keyName;
 
   private DatabaseAdminClient dbAdminClient;
   private InstanceAdminClient instanceAdminClient;
@@ -101,6 +107,10 @@ public class ITBackupTest {
   @BeforeClass
   public static void doNotRunOnEmulator() {
     assumeFalse("backups are not supported on the emulator", isUsingEmulator());
+    keyName = System.getProperty(KMS_KEY_NAME_PROPERTY);
+    Preconditions.checkNotNull(
+        keyName,
+        "Key name is null, please set a key to be used for this test. The necessary permissions should be grant to the spanner service account according to the CMEK user guide.");
   }
 
   @Before
@@ -199,13 +209,18 @@ private void waitForDbOperations(String backupId) throws InterruptedException {
   @Test
   public void testBackups() throws InterruptedException, ExecutionException {
     // Create two test databases in parallel.
-    String db1Id = testHelper.getUniqueDatabaseId() + "_db1";
+    final String db1Id = testHelper.getUniqueDatabaseId() + "_db1";
+    final Database sourceDatabase1 =
+        dbAdminClient
+            .newDatabaseBuilder(DatabaseId.of(projectId, instanceId, db1Id))
+            .setEncryptionConfig(EncryptionConfigs.customerManagedEncryption(keyName))
+            .build();
     logger.info(String.format("Creating test database %s", db1Id));
     OperationFuture dbOp1 =
         dbAdminClient.createDatabase(
-            testHelper.getInstanceId().getInstance(),
-            db1Id,
-            Arrays.asList("CREATE TABLE FOO (ID INT64, NAME STRING(100)) PRIMARY KEY (ID)"));
+            sourceDatabase1,
+            Collections.singletonList(
+                "CREATE TABLE FOO (ID INT64, NAME STRING(100)) PRIMARY KEY (ID)"));
     String db2Id = testHelper.getUniqueDatabaseId() + "_db2";
     logger.info(String.format("Creating test database %s", db2Id));
     OperationFuture dbOp2 =
@@ -229,6 +244,9 @@ public void testBackups() throws InterruptedException, ExecutionException {
                 .to("TEST")
                 .build()));
 
+    // Verifies that the database encryption has been properly set
+    testDatabaseEncryption(db1);
+
     // Create two backups in parallel.
     String backupId1 = testHelper.getUniqueBackupId() + "_bck1";
     String backupId2 = testHelper.getUniqueBackupId() + "_bck2";
@@ -236,12 +254,14 @@ public void testBackups() throws InterruptedException, ExecutionException {
     Timestamp versionTime = getCurrentTimestamp(client);
     logger.info(String.format("Creating backups %s and %s in parallel", backupId1, backupId2));
     // This backup has the version time specified as the server's current timestamp
+    // This backup is encrypted with a customer managed key
     final Backup backupToCreate1 =
         dbAdminClient
             .newBackupBuilder(BackupId.of(projectId, instanceId, backupId1))
             .setDatabase(db1.getId())
             .setExpireTime(expireTime)
             .setVersionTime(versionTime)
+            .setEncryptionConfig(EncryptionConfigs.customerManagedEncryption(keyName))
             .build();
     // This backup has no version time specified
     final Backup backupToCreate2 =
@@ -287,12 +307,14 @@ public void testBackups() throws InterruptedException, ExecutionException {
             "Backup2 still not finished. Test is giving up waiting for it.");
       }
       logger.info("Long-running operations finished. Getting backups by id.");
-      backup1 = dbAdminClient.getBackup(this.instance.getId().getInstance(), backupId1);
-      backup2 = dbAdminClient.getBackup(this.instance.getId().getInstance(), backupId2);
+      backup1 = dbAdminClient.getBackup(instance.getId().getInstance(), backupId1);
+      backup2 = dbAdminClient.getBackup(instance.getId().getInstance(), backupId2);
     }
 
     // Verifies that backup version time is the specified one
     testBackupVersionTime(backup1, versionTime);
+    // Verifies that backup encryption has been properly set
+    testBackupEncryption(backup1);
 
     // Insert some more data into db2 to get a timestamp from the server.
     Timestamp commitTs =
@@ -308,7 +330,7 @@ public void testBackups() throws InterruptedException, ExecutionException {
     // Test listing operations.
     // List all backups.
     logger.info("Listing all backups");
-    assertThat(this.instance.listBackups().iterateAll()).containsAtLeast(backup1, backup2);
+    assertThat(instance.listBackups().iterateAll()).containsAtLeast(backup1, backup2);
     // List all backups whose names contain 'bck1'.
     logger.info("Listing backups with name bck1");
     assertThat(
@@ -425,6 +447,20 @@ private void testBackupVersionTime(Backup backup, Timestamp versionTime) {
     logger.info("Done verifying backup version time for " + backup.getId());
   }
 
+  private void testDatabaseEncryption(Database database) {
+    logger.info("Verifying database encryption for " + database.getId());
+    assertThat(database.getEncryptionConfig()).isNotNull();
+    assertThat(database.getEncryptionConfig().getKmsKeyName()).isEqualTo(keyName);
+    logger.info("Done verifying database encryption for " + database.getId());
+  }
+
+  private void testBackupEncryption(Backup backup) {
+    logger.info("Verifying backup encryption for " + backup.getId());
+    assertThat(backup.getEncryptionInfo()).isNotNull();
+    assertThat(backup.getEncryptionInfo().getKmsKeyVersion()).isNotNull();
+    logger.info("Done verifying backup encryption for " + backup.getId());
+  }
+
   private void testMetadata(
       OperationFuture op1,
       OperationFuture op2,
@@ -583,15 +619,20 @@ private void testRestore(
     // Restore the backup to a new database.
     String restoredDb = testHelper.getUniqueDatabaseId();
     String restoreOperationName;
-    OperationFuture restoreOp;
+    OperationFuture restoreOperation;
     int attempts = 0;
     while (true) {
       try {
         logger.info(
             String.format(
                 "Restoring backup %s to database %s", backup.getId().getBackup(), restoredDb));
-        restoreOp = backup.restore(DatabaseId.of(testHelper.getInstanceId(), restoredDb));
-        restoreOperationName = restoreOp.getName();
+        final Restore restore =
+            dbAdminClient
+                .newRestoreBuilder(backup.getId(), DatabaseId.of(projectId, instanceId, restoredDb))
+                .setEncryptionConfig(EncryptionConfigs.customerManagedEncryption(keyName))
+                .build();
+        restoreOperation = dbAdminClient.restoreDatabase(restore);
+        restoreOperationName = restoreOperation.getName();
         break;
       } catch (ExecutionException e) {
         if (e.getCause() instanceof FailedPreconditionException
@@ -617,7 +658,7 @@ private void testRestore(
     }
     databases.add(restoredDb);
     logger.info(String.format("Restore operation %s running", restoreOperationName));
-    RestoreDatabaseMetadata metadata = restoreOp.getMetadata().get();
+    RestoreDatabaseMetadata metadata = restoreOperation.getMetadata().get();
     assertThat(metadata.getBackupInfo().getBackup()).isEqualTo(backup.getId().getName());
     assertThat(metadata.getSourceType()).isEqualTo(RestoreSourceType.BACKUP);
     assertThat(metadata.getName())
@@ -630,7 +671,7 @@ private void testRestore(
     //    verifyRestoreOperations(backupOp.getName(), restoreOperationName);
 
     // Wait until the restore operation has finished successfully.
-    Database database = restoreOp.get();
+    Database database = restoreOperation.get();
     assertThat(database.getId().getDatabase()).isEqualTo(restoredDb);
 
     // Reloads the database
@@ -639,6 +680,7 @@ private void testRestore(
             Timestamp.fromProto(
                 reloadedDatabase.getProto().getRestoreInfo().getBackupInfo().getVersionTime()))
         .isEqualTo(versionTime);
+    testDatabaseEncryption(reloadedDatabase);
 
     // Restoring the backup to an existing database should fail.
     try {