### Question:
AbstractJdbcDataSetProvider implements DataSetProvider<T> { @Nonnull @VisibleForTesting JdbcHighWaterMark createHighWaterMark(@Nonnull final String highWaterMarkKey, @Nonnull final KyloCatalogClient<T> client) { final JdbcHighWaterMark highWaterMark = new JdbcHighWaterMark(highWaterMarkKey, client); highWaterMark.setFormatter(new LongToDateTime()); final String value = client.getHighWaterMarks().get(highWaterMarkKey); if (value != null) { try { highWaterMark.accumulate(ISODateTimeFormat.dateTimeParser().withZoneUTC().parseMillis(value)); } catch (final IllegalArgumentException e) { throw new KyloCatalogException("Invalid value for high water mark " + highWaterMarkKey + ": " + value, e); } } return highWaterMark; }  @Override boolean supportsFormat(@Nonnull final String source); @Nonnull @Override final T read(@Nonnull final KyloCatalogClient<T> client, @Nonnull final DataSetOptions options); @Override final void write(@Nonnull final KyloCatalogClient<T> client, @Nonnull final DataSetOptions options, @Nonnull final T dataSet);  }### Answer:
@Test public void createHighWaterMark() { final KyloCatalogClient client = Mockito.mock(KyloCatalogClient.class); final MockJdbcDataSetProvider provider = new MockJdbcDataSetProvider(); final JdbcHighWaterMark highWaterMark = provider.createHighWaterMark("mockWaterMark", client); Assert.assertEquals("mockWaterMark", highWaterMark.getName()); Assert.assertNull("Expected initial value to be null", highWaterMark.getValue()); highWaterMark.accumulate(1525014303000L); Mockito.verify(client).setHighWaterMarks(Collections.singletonMap("mockWaterMark", "2018-04-29T15:05:03")); }
@Test public void createHighWaterMarkWithExisting() { final KyloCatalogClient client = Mockito.mock(KyloCatalogClient.class); Mockito.when(client.getHighWaterMarks()).thenReturn(Collections.singletonMap("mockWaterMark", "2018-04-29T15:05:03")); final MockJdbcDataSetProvider provider = new MockJdbcDataSetProvider(); final JdbcHighWaterMark highWaterMark = provider.createHighWaterMark("mockWaterMark", client); Assert.assertEquals("mockWaterMark", highWaterMark.getName()); Assert.assertEquals(new Long(1525014303000L), highWaterMark.getValue()); Mockito.reset(client); highWaterMark.accumulate(86400000L); Mockito.verifyZeroInteractions(client); highWaterMark.accumulate(1532528828000L); Mockito.verify(client).setHighWaterMarks(Collections.singletonMap("mockWaterMark", "2018-07-25T14:27:08")); }