CryptoSerializer implements Serializer<T> { @Override public byte[] serialize(String topic, Headers headers, T data) { byte[] serializedData = rawSerializer.serialize(topic, headers, data); if (serializedData == null) { return null; } Header keyReferenceHeader = headers.lastHeader(KEY_REF_HEADER); return encrypt(serializedData, keyReferenceHeader == null ? null : keyReferenceHeader.value()); } CryptoSerializer(Serializer<? super T> rawSerializer, Encryptor encryptor, ThreadLocal<byte[]> keyRefHolder); @Override byte[] serialize(String topic, Headers headers, T data); @Override byte[] serialize(String topic, T data); @Override void configure(Map<String, ?> configs, boolean isKey); @Override void close();  }
@Test public void testSerializeWhenKeyRefHeaderIsNotSet() { final String clearPayload = "clear serialized value"; final int clearPayloadSize = toByteArray(clearPayload).length; Headers headers = new RecordHeaders(); given(rawSerializer.serialize("topic1", headers, "final value")) .willReturn(toByteArray(clearPayload)); byte[] result = cryptoSerializer.serialize("topic1", headers, "final value"); assertThat(result).isEqualTo("clear serialized value".getBytes(StandardCharsets.UTF_8)); verifyZeroInteractions(encryptor); }
@Test public void testSerializeWhenKeyRefHeaderIsSetToNull() { final String clearPayload = "clear serialized value"; final int clearPayloadSize = toByteArray(clearPayload).length; Headers headers = new RecordHeaders().add(KafkaCryptoConstants.KEY_REF_HEADER, null); given(rawSerializer.serialize("topic1", headers, "final value")) .willReturn(toByteArray(clearPayload)); byte[] result = cryptoSerializer.serialize("topic1", headers, "final value"); assertThat(result).isEqualTo("clear serialized value".getBytes(StandardCharsets.UTF_8)); verifyZeroInteractions(encryptor); }