@Override
        public void process(final Record<Integer, String> record) {
            // add record to store
            if (DataTracker.DELETE_VALUE_KEYWORD.equals(record.value())) {
                // special value "delete" is interpreted as a delete() call from
                // VersionedStoreContentCheckerProcessor but we do not support it here
                throw new IllegalArgumentException("Using 'delete' keyword for "
                    + "TimestampedStoreContentCheckerProcessor will result in the record "
                    + "timestamp being ignored. Use regular put with null value instead.");
            }
            final ValueAndTimestamp<String> valueAndTimestamp = ValueAndTimestamp.make(record.value(), record.timestamp());
            store.put(record.key(), valueAndTimestamp);
            data.put(record.key(), Optional.ofNullable(valueAndTimestamp));

            // check expected contents of store, and signal completion by writing
            // number of failures to downstream
            final int failedChecks = checkStoreContents();
            context.forward(record.withValue(failedChecks));
        }
Same as {@link VersionedStoreContentCheckerProcessor} but for timestamped stores instead,
for use in validating the manual upgrade path from non-versioned to versioned stores.
streams/src/test/java/org/apache/kafka/streams/integration/VersionedKeyValueStoreIntegrationTest.java