@Override
        public void process(final Record<Integer, String> record) {
            if (writeToStore) {
                // add record to store. special value "delete" is interpreted as a delete() call,
                // in contrast to null value, which is a tombstone inserted via put()
                if (DataTracker.DELETE_VALUE_KEYWORD.equals(record.value())) {
                    store.delete(record.key(), record.timestamp());
                } else {
                    store.put(record.key(), record.value(), record.timestamp());
                }
                data.add(record.key(), record.timestamp(), record.value());
            }

            // check expected contents of store, and signal completion by writing
            // number of failures to downstream
            final int failedChecks = checkStoreContents();
            context.forward(record.withValue(failedChecks));
        }
@param writeToStore whether or not this processor should write to the store
@param initialData  expected store contents which have already been inserted from
                    outside of this processor
streams/src/test/java/org/apache/kafka/streams/integration/VersionedKeyValueStoreIntegrationTest.java