@SuppressWarnings("unchecked")
    @Override
    public void update(final ConsumerRecord<byte[], byte[]> record) {
        final RecordDeserializer sourceNodeAndDeserializer = deserializers.get(record.topic());
        final ConsumerRecord<Object, Object> deserialized = sourceNodeAndDeserializer.deserialize(processorContext, record);

        if (deserialized != null) {
            final ProcessorRecordContext recordContext =
                new ProcessorRecordContext(
                    deserialized.timestamp(),
                    deserialized.offset(),
                    deserialized.partition(),
                    deserialized.topic(),
                    deserialized.headers(),
                    record);
            processorContext.setRecordContext(recordContext);
            processorContext.setCurrentNode(sourceNodeAndDeserializer.sourceNode());
            final Record<Object, Object> toProcess = new Record<>(
                deserialized.key(),
                deserialized.value(),
                processorContext.timestamp(),
                processorContext.headers()
            );
            ((SourceNode<Object, Object>) sourceNodeAndDeserializer.sourceNode()).process(toProcess);
        }

        offsets.put(new TopicPartition(record.topic(), record.partition()), record.offset() + 1);
    }
@throws IllegalStateException If store gets registered after initialized is already finished
@throws StreamsException      If the store's change log does not contain the partition
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java