CryptoDeserializer implements Deserializer<T> { @Override public T deserialize(String topic, Headers headers, byte[] data) { if (data == null) { return null; } DecryptedDataWithKeyRef decryptedDataWithKeyRef = decrypt(data); T deserializedValue = rawDeserializer.deserialize(topic, headers, decryptedDataWithKeyRef.decryptedData); headers.add(KEY_REF_HEADER, decryptedDataWithKeyRef.keyRef); return deserializedValue; } CryptoDeserializer(Deserializer<? extends T> rawDeserializer, Decryptor decryptor); @Override T deserialize(String topic, Headers headers, byte[] data); @Override void configure(Map<String, ?> configs, boolean isKey); @Override T deserialize(String topic, byte[] data); @Override void close();  }
@Test public void testDeserializeWhenKeyRefIsNotSet() { Headers headers = new RecordHeaders(); byte[] clearValue = "clearValue".getBytes(StandardCharsets.UTF_8); ByteBuffer rawValue = ByteBuffer.allocate(KafkaCryptoConstants.ENCRYPTED_PREFIX.length + Integer.BYTES + clearValue.length); rawValue.put(KafkaCryptoConstants.ENCRYPTED_PREFIX); rawValue.putInt(0); rawValue.put(clearValue); given(rawDeserializer.deserialize("topic1", headers, clearValue)).willReturn("deserialized value"); String value = cryptoDeserializer.deserialize("topic1", headers, rawValue.array()); assertThat(value).isEqualTo("deserialized value"); assertThat(headers.lastHeader(KafkaCryptoConstants.KEY_REF_HEADER).value()).isNull(); verifyZeroInteractions(decryptor); }
@Test public void testDeserializeWhenNoEncryptionStructure() { Headers headers = new RecordHeaders(); byte[] clearValue = "clearValue".getBytes(StandardCharsets.UTF_8); given(rawDeserializer.deserialize("topic1", headers, clearValue)).willReturn("deserialized value"); String value = cryptoDeserializer.deserialize("topic1", headers, clearValue); assertThat(value).isEqualTo("deserialized value"); assertThat(headers.lastHeader(KafkaCryptoConstants.KEY_REF_HEADER).value()).isNull(); verifyZeroInteractions(decryptor); }