@Override
    public void openExisting(final ProcessorContext context, final long streamTime) {
        metricsRecorder.init(ProcessorContextUtils.metricsImpl(context), context.taskId());
        super.openExisting(context, streamTime);
    }
@Test
    public void shouldUpdateSegmentFileNameFromOldDateFormatToNewFormat() throws Exception {
        final long segmentInterval = 60_000L; // the old segment file's naming system maxes out at 1 minute granularity.

        segments = new TimestampedSegments(storeName, METRICS_SCOPE, NUM_SEGMENTS * segmentInterval, segmentInterval);

        final String storeDirectoryPath = stateDirectory.getAbsolutePath() + File.separator + storeName;
        final File storeDirectory = new File(storeDirectoryPath);
        //noinspection ResultOfMethodCallIgnored
        storeDirectory.mkdirs();

        final SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmm");
        formatter.setTimeZone(new SimpleTimeZone(0, "UTC"));

        for (int segmentId = 0; segmentId < NUM_SEGMENTS; ++segmentId) {
            final File oldSegment = new File(storeDirectoryPath + File.separator + storeName + "-" + formatter.format(new Date(segmentId * segmentInterval)));
            //noinspection ResultOfMethodCallIgnored
            Files.createFile(oldSegment.toPath());
        }

        segments.openExisting(context, -1L);

        for (int segmentId = 0; segmentId < NUM_SEGMENTS; ++segmentId) {
            final String segmentName = storeName + "." + (long) segmentId * segmentInterval;
            final File newSegment = new File(storeDirectoryPath + File.separator + segmentName);
            assertTrue(Files.exists(newSegment.toPath()));
        }
    }