Allow TransformedRecordSerializerJCE.Builder to be reused

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/TransformedRecordSerializerJCE.java

@@ -304,13 +304,19 @@ public Builder<M> clearKeyManager() {
          */
         @Override
         public TransformedRecordSerializerJCE<M> build() {
-            if (keyManager == null) {
-                if (encryptionKey != null) {
-                    keyManager = new FixedZeroKeyManager(encryptionKey, cipherName, secureRandom);
-                } else if (encryptWhenSerializing) {
-                    throw new RecordCoreArgumentException("cannot encrypt when serializing if encryption key is not set");
-                }
-            } else {
+            return new TransformedRecordSerializerJCE<>(
+                    inner,
+                    compressWhenSerializing,
+                    compressionLevel,
+                    encryptWhenSerializing,
+                    writeValidationRatio,
+                    resolveKeyManager()
+            );
+        }
+
+        @Nullable
+        private SerializationKeyManager resolveKeyManager() {
+            if (this.keyManager != null) {
                 if (encryptionKey != null) {
                     throw new RecordCoreArgumentException("cannot specify both key manager and encryption key");
                 }
@@ -320,17 +326,20 @@ public TransformedRecordSerializerJCE<M> build() {
                 if (secureRandom != null) {
                     throw new RecordCoreArgumentException("cannot specify both key manager and secure random");
                 }
+
+                return keyManager;
+            }
+
+            if (encryptionKey != null) {
+                return new FixedZeroKeyManager(encryptionKey, cipherName, secureRandom);
             }
-            return new TransformedRecordSerializerJCE<>(
-                    inner,
-                    compressWhenSerializing,
-                    compressionLevel,
-                    encryptWhenSerializing,
-                    writeValidationRatio,
-                    keyManager
-            );
-        }
 
+            if (!encryptWhenSerializing) {
+                return null;
+            }
+
+            throw new RecordCoreArgumentException("cannot encrypt when serializing if encryption key is not set");
+        }
     }
 
 }

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/common/TransformedRecordSerializerTest.java

@@ -72,6 +72,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.not;
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -772,6 +773,25 @@ void invalidKeyManagerBuilder() throws Exception {
         assertThat(e.getMessage(), containsString("cannot specify both key manager and encryption key"));
     }
 
+    @Test
+    void reuseBuilder() throws Exception {
+        RollingTestKeyManager keyManager = new RollingTestKeyManager(0);
+        TransformedRecordSerializerJCE.Builder<Message> builderWithEncryptionKey = TransformedRecordSerializerJCE
+                .newDefaultBuilder()
+                .setEncryptionKey(keyManager.getKey(keyManager.getSerializationKey()));
+        TransformedRecordSerializerJCE.Builder<Message> builderWithKeyManager = TransformedRecordSerializerJCE
+                .newDefaultBuilder()
+                .setKeyManager(keyManager);
+
+        List.of(builderWithEncryptionKey, builderWithKeyManager).forEach((builder) -> {
+            builder.setEncryptWhenSerializing(false);
+            assertDoesNotThrow(builder::build);
+
+            builder.setEncryptWhenSerializing(true);
+            assertDoesNotThrow(builder::build);
+        });
+    }
+
     private boolean isCompressed(byte[] serialized) {
         byte headerByte = serialized[0];
         return headerByte == TransformedRecordSerializerPrefix.PREFIX_COMPRESSED ||