RetryTimeout implements BackgroundFunction<PubSubMessage> { @Override public void accept(PubSubMessage message, Context context) { ZonedDateTime utcNow = ZonedDateTime.now(ZoneOffset.UTC); ZonedDateTime timestamp = ZonedDateTime.parse(context.timestamp()); long eventAge = Duration.between(timestamp, utcNow).toMillis(); if (eventAge > MAX_EVENT_AGE) { logger.info(String.format("Dropping event with timestamp %s.", timestamp)); return; } logger.info(String.format("Processing event with timestamp %s.", timestamp)); }  @Override void accept(PubSubMessage message, Context context);  }
@Test public void retryTimeout_handlesRetryMsg() { ZonedDateTime timestamp = ZonedDateTime.now(ZoneOffset.UTC); Context mockContext = mock(Context.class); when(mockContext.timestamp()).thenReturn(timestamp.toString()); PubSubMessage pubsubMessage = new PubSubMessage(); new RetryTimeout().accept(pubsubMessage, mockContext); String logMessage = LOG_HANDLER.getStoredLogRecords().get(0).getMessage(); assertThat(logMessage).contains("Processing event with timestamp " + timestamp); }
@Test public void retryTimeout_handlesStopMsg() { ZonedDateTime timestamp = ZonedDateTime.ofInstant(Instant.ofEpochMilli(0), ZoneOffset.UTC); Context mockContext = mock(Context.class); when(mockContext.timestamp()).thenReturn(timestamp.toString()); PubSubMessage pubsubMessage = new PubSubMessage(); new RetryTimeout().accept(pubsubMessage, mockContext); String logMessage = LOG_HANDLER.getStoredLogRecords().get(0).getMessage(); assertThat(logMessage).contains("Dropping event with timestamp " + timestamp); }