@Test public void shouldReturnNewConsumerRecordWithDeserializedValueWhenNoExceptions() throws Exception { final SourceNodeRecordDeserializer recordDeserializer = new SourceNodeRecordDeserializer( new TheSourceNode(false, false, "key", "value")); final ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(rawRecord); assertEquals(rawRecord.topic(), record.topic()); assertEquals(rawRecord.partition(), record.partition()); assertEquals(rawRecord.offset(), record.offset()); assertEquals(rawRecord.checksum(), record.checksum()); assertEquals("key", record.key()); assertEquals("value", record.value()); assertEquals(rawRecord.timestamp(), record.timestamp()); assertEquals(TimestampType.CREATE_TIME, record.timestampType()); }
@Override public ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord) { final Object key; try { key = sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize key for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } final Object value; try { value = sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize value for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } return new ConsumerRecord<>(rawRecord.topic(), rawRecord.partition(), rawRecord.offset(), rawRecord.timestamp(), TimestampType.CREATE_TIME, rawRecord.checksum(), rawRecord.serializedKeySize(), rawRecord.serializedValueSize(), key, value); }
SourceNodeRecordDeserializer implements RecordDeserializer { @Override public ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord) { final Object key; try { key = sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize key for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } final Object value; try { value = sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize value for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } return new ConsumerRecord<>(rawRecord.topic(), rawRecord.partition(), rawRecord.offset(), rawRecord.timestamp(), TimestampType.CREATE_TIME, rawRecord.checksum(), rawRecord.serializedKeySize(), rawRecord.serializedValueSize(), key, value); } }
SourceNodeRecordDeserializer implements RecordDeserializer { @Override public ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord) { final Object key; try { key = sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize key for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } final Object value; try { value = sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize value for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } return new ConsumerRecord<>(rawRecord.topic(), rawRecord.partition(), rawRecord.offset(), rawRecord.timestamp(), TimestampType.CREATE_TIME, rawRecord.checksum(), rawRecord.serializedKeySize(), rawRecord.serializedValueSize(), key, value); } SourceNodeRecordDeserializer(final SourceNode sourceNode); }
SourceNodeRecordDeserializer implements RecordDeserializer { @Override public ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord) { final Object key; try { key = sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize key for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } final Object value; try { value = sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize value for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } return new ConsumerRecord<>(rawRecord.topic(), rawRecord.partition(), rawRecord.offset(), rawRecord.timestamp(), TimestampType.CREATE_TIME, rawRecord.checksum(), rawRecord.serializedKeySize(), rawRecord.serializedValueSize(), key, value); } SourceNodeRecordDeserializer(final SourceNode sourceNode); @Override ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord); }
SourceNodeRecordDeserializer implements RecordDeserializer { @Override public ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord) { final Object key; try { key = sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize key for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } final Object value; try { value = sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()); } catch (Exception e) { throw new StreamsException(format("Failed to deserialize value for record. topic=%s, partition=%d, offset=%d", rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e); } return new ConsumerRecord<>(rawRecord.topic(), rawRecord.partition(), rawRecord.offset(), rawRecord.timestamp(), TimestampType.CREATE_TIME, rawRecord.checksum(), rawRecord.serializedKeySize(), rawRecord.serializedValueSize(), key, value); } SourceNodeRecordDeserializer(final SourceNode sourceNode); @Override ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord);  }