### Question:
KafkaProtobufDeserializer implements Deserializer<T> { @Override public T deserialize( String topic, byte[] data ) { try { return parser.parseFrom( data ); } catch ( InvalidProtocolBufferException e ) { throw new SerializationException( "Error deserializing from Protobuf message", e ); } } KafkaProtobufDeserializer( Parser<T> parser ); @Override void configure( Map<String, ?> configs, boolean isKey ); @Override T deserialize( String topic, byte[] data ); @Override void close();  }### Answer:
@Test public void testDeserialize() throws InvalidProtocolBufferException { Parser<MessageLite> parser = mock( Parser.class ); KafkaProtobufDeserializer<MessageLite> deserializer = new KafkaProtobufDeserializer( parser ); deserializer.deserialize( "topic", "data".getBytes() ); verify( parser ).parseFrom( "data".getBytes() ); deserializer.configure( null, false ); deserializer.close(); verifyNoMoreInteractions( parser ); }
@Test( expected = SerializationException.class ) public void testInvalidProtocolBufferException() throws InvalidProtocolBufferException { Parser<MessageLite> parser = mock( Parser.class ); doThrow( new InvalidProtocolBufferException( "" ) ).when( parser ).parseFrom( (byte[]) any() ); KafkaProtobufDeserializer<MessageLite> deserializer = new KafkaProtobufDeserializer( parser ); deserializer.deserialize( "topic", "data".getBytes() ); }