AbstractKafkaConnector implements Connector, DiagnosticsAware { @VisibleForTesting long getThreadDelayTimeInSecond(OffsetDateTime now, int daemonThreadIntervalSeconds) { long truncatedTimestamp = now.truncatedTo(ChronoUnit.HOURS).toEpochSecond(); long minDelay = Math.min(MIN_DAEMON_THREAD_STARTUP_DELAY.getSeconds(), daemonThreadIntervalSeconds); while ((truncatedTimestamp - now.toEpochSecond()) < minDelay) { truncatedTimestamp += daemonThreadIntervalSeconds; } return truncatedTimestamp - now.toEpochSecond(); } AbstractKafkaConnector(String connectorName, Properties config, GroupIdConstructor groupIdConstructor,
      String clusterName, Logger logger); @Override synchronized void onAssignmentChange(List<DatastreamTask> tasks); Thread createTaskThread(AbstractKafkaBasedConnectorTask task); @Override void start(CheckpointProvider checkpointProvider); @Override void stop(); @Override void validateUpdateDatastreams(List<Datastream> datastreams, List<Datastream> allDatastreams); @Override String process(String query); @Override String reduce(String query, Map<String, String> responses); @Override boolean isDatastreamUpdateTypeSupported(Datastream datastream, DatastreamConstants.UpdateType updateType); @Override List<BrooklinMetricInfo> getMetricInfos(); static final String IS_GROUP_ID_HASHING_ENABLED; }
@Test public void testCalculateThreadStartDelay() { Properties props = new Properties(); props.setProperty("daemonThreadIntervalInSeconds", "30"); TestKafkaConnector connector = new TestKafkaConnector(false, props); Duration threadDelay = Duration.ofMinutes(5); Duration minDelay = AbstractKafkaConnector.MIN_DAEMON_THREAD_STARTUP_DELAY; Duration testPeriod = Duration.ofDays(1).plusHours(2); OffsetDateTime start = OffsetDateTime.now(ZoneOffset.UTC).truncatedTo(ChronoUnit.SECONDS); OffsetDateTime end = start.plus(testPeriod); for (OffsetDateTime now = start; now.isBefore(end); now = now.plusSeconds(1)) { OffsetDateTime currentTime = now; OffsetDateTime expectedTime = Stream.iterate(now.truncatedTo(ChronoUnit.HOURS), time -> time.plus(threadDelay)) .filter(runTime -> runTime.isAfter(currentTime) && !runTime.isBefore(currentTime.plus(minDelay))) .findFirst() .orElseThrow(() -> new RuntimeException(String.format("Failed to find next expected thread runtime with " + "base time of %s, thread delay %s, min delay %s, and current time %s", start, threadDelay, minDelay, currentTime))) .truncatedTo(ChronoUnit.SECONDS); long actualDelaySeconds = connector.getThreadDelayTimeInSecond(now, (int) threadDelay.getSeconds()); OffsetDateTime actualTime = now.plusSeconds(actualDelaySeconds); Assert.assertEquals(actualTime, expectedTime, String.format("Actual = %s, Expected = %s, with base time of %s, " + "thread delay %s, min delay %s, and current time %s", actualTime, expectedTime, start, threadDelay, minDelay, now)); } }