@Test(expected = StreamsException.class) public void shouldThrowStreamsExceptionWhenValueDeserializationFails() throws Exception { final byte[] value = Serdes.Long().serializer().serialize("foo", 1L); final List<ConsumerRecord<byte[], byte[]>> records = Collections.singletonList( new ConsumerRecord<>("topic", 1, 1, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, recordKey, value)); queue.addRawRecords(records); }
public int addRawRecords(Iterable<ConsumerRecord<byte[], byte[]>> rawRecords) { for (ConsumerRecord<byte[], byte[]> rawRecord : rawRecords) { ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(rawRecord); long timestamp = timestampExtractor.extract(record, timeTracker.get()); log.trace("Source node {} extracted timestamp {} for record {}", source.name(), timestamp, record); if (timestamp < 0) { continue; } StampedRecord stampedRecord = new StampedRecord(record, timestamp); fifoQueue.addLast(stampedRecord); timeTracker.addElement(stampedRecord); } long timestamp = timeTracker.get(); if (timestamp > partitionTime) partitionTime = timestamp; return size(); }
RecordQueue { public int addRawRecords(Iterable<ConsumerRecord<byte[], byte[]>> rawRecords) { for (ConsumerRecord<byte[], byte[]> rawRecord : rawRecords) { ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(rawRecord); long timestamp = timestampExtractor.extract(record, timeTracker.get()); log.trace("Source node {} extracted timestamp {} for record {}", source.name(), timestamp, record); if (timestamp < 0) { continue; } StampedRecord stampedRecord = new StampedRecord(record, timestamp); fifoQueue.addLast(stampedRecord); timeTracker.addElement(stampedRecord); } long timestamp = timeTracker.get(); if (timestamp > partitionTime) partitionTime = timestamp; return size(); } }
RecordQueue { public int addRawRecords(Iterable<ConsumerRecord<byte[], byte[]>> rawRecords) { for (ConsumerRecord<byte[], byte[]> rawRecord : rawRecords) { ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(rawRecord); long timestamp = timestampExtractor.extract(record, timeTracker.get()); log.trace("Source node {} extracted timestamp {} for record {}", source.name(), timestamp, record); if (timestamp < 0) { continue; } StampedRecord stampedRecord = new StampedRecord(record, timestamp); fifoQueue.addLast(stampedRecord); timeTracker.addElement(stampedRecord); } long timestamp = timeTracker.get(); if (timestamp > partitionTime) partitionTime = timestamp; return size(); } RecordQueue(final TopicPartition partition,
                final SourceNode source,
                final TimestampExtractor timestampExtractor); }
RecordQueue { public int addRawRecords(Iterable<ConsumerRecord<byte[], byte[]>> rawRecords) { for (ConsumerRecord<byte[], byte[]> rawRecord : rawRecords) { ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(rawRecord); long timestamp = timestampExtractor.extract(record, timeTracker.get()); log.trace("Source node {} extracted timestamp {} for record {}", source.name(), timestamp, record); if (timestamp < 0) { continue; } StampedRecord stampedRecord = new StampedRecord(record, timestamp); fifoQueue.addLast(stampedRecord); timeTracker.addElement(stampedRecord); } long timestamp = timeTracker.get(); if (timestamp > partitionTime) partitionTime = timestamp; return size(); } RecordQueue(final TopicPartition partition,
                final SourceNode source,
                final TimestampExtractor timestampExtractor); SourceNode source(); TopicPartition partition(); int addRawRecords(Iterable<ConsumerRecord<byte[], byte[]>> rawRecords); StampedRecord poll(); int size(); boolean isEmpty(); long timestamp(); void clear(); }
RecordQueue { public int addRawRecords(Iterable<ConsumerRecord<byte[], byte[]>> rawRecords) { for (ConsumerRecord<byte[], byte[]> rawRecord : rawRecords) { ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(rawRecord); long timestamp = timestampExtractor.extract(record, timeTracker.get()); log.trace("Source node {} extracted timestamp {} for record {}", source.name(), timestamp, record); if (timestamp < 0) { continue; } StampedRecord stampedRecord = new StampedRecord(record, timestamp); fifoQueue.addLast(stampedRecord); timeTracker.addElement(stampedRecord); } long timestamp = timeTracker.get(); if (timestamp > partitionTime) partitionTime = timestamp; return size(); } RecordQueue(final TopicPartition partition,
                final SourceNode source,
                final TimestampExtractor timestampExtractor); SourceNode source(); TopicPartition partition(); int addRawRecords(Iterable<ConsumerRecord<byte[], byte[]>> rawRecords); StampedRecord poll(); int size(); boolean isEmpty(); long timestamp(); void clear();  }