### Question:
AbstractClientCallbacksForJDBC implements WaltzClientCallbacks { protected void onErrorGettingClientHighWaterMark(int attempts, SQLException exception) { } AbstractClientCallbacksForJDBC(DataSource dataSource, String clientHighWaterMarkTableName); @Override long getClientHighWaterMark(int partitionId); @Override void applyTransaction(Transaction transaction);  }### Answer:
@Test public void testOnErrorGettingClientHighWaterMark() throws Exception { TestDataSource dataSource = new TestDataSource("TEST_EXCEPTION_IN_GET_CLIENT_HIGH_WATER_MARK"); Connection connection = dataSource.getConnection(); dropSchema(connection); createSchema(connection); Map<Integer, MockServerPartition> serverPartitions = MockServerPartition.create(1); MockDriver mockDriver1 = new MockDriver(1, serverPartitions); WaltzClientConfig config1 = new WaltzClientConfig(new Properties()); config1.setObject(WaltzClientConfig.MOCK_DRIVER, mockDriver1); WaltzClientCallbacks callbacks1 = new AbstractClientCallbacksForJDBC(dataSource, CLIENT_HIGH_WATER_MARK_TABLE_NAME) { private int expectedAttempts = 0; @Override protected void onErrorGettingClientHighWaterMark(int attempts, SQLException exception) { expectedAttempts++; assertEquals(expectedAttempts, attempts); if (attempts == 5) { throw new RuntimeException("no more attempts", exception); } else if (attempts > 5) { fail(); } } @Override protected void applyTransaction(Transaction transaction, Connection connection) throws SQLException { } @Override public void uncaughtException(int partitionId, long transactionId, Throwable exception) { } }; dataSource.makeDbDown(); try { new WaltzClient(callbacks1, config1); fail(); } catch (RuntimeException ex) { assertEquals("no more attempts", ex.getMessage()); assertTrue(ex.getCause() instanceof SQLException); } }