@Override
  public void start(
      final KsqlModuleType moduleType,
      final Properties ksqlProperties) {
    final BaseSupportConfig ksqlVersionCheckerConfig =
        new PhoneHomeConfig(ksqlProperties, "ksql");

    if (!ksqlVersionCheckerConfig.isProactiveSupportEnabled()) {
      log.warn(legalDisclaimerProactiveSupportDisabled());
      return;
    }
    try {
      final KsqlVersionChecker ksqlVersionChecker = versionCheckerFactory.create(
          ksqlVersionCheckerConfig,
          moduleType,
          enableSettlingTime,
          this::isActive
      );

      ksqlVersionChecker.init();
      ksqlVersionChecker.setUncaughtExceptionHandler((t, e)
          -> log.error("Uncaught exception in thread '{}':", t.getName(), e));
      ksqlVersionChecker.start();
      final long reportIntervalMs = ksqlVersionCheckerConfig.getReportIntervalMs();
      final long reportIntervalHours = reportIntervalMs / (60 * 60 * 1000);
      // We log at WARN level to increase the visibility of this information.
      log.warn(legalDisclaimerProactiveSupportEnabled(reportIntervalHours));

    } catch (final Exception e) {
      // We catch any exceptions to prevent collateral damage to the more important broker
      // threads that are running in the same JVM.
      log.error("Failed to start KsqlVersionCheckerAgent: {}", e.getMessage());
    }
  }
@Test(expected = Exception.class)
  public void shouldFailIfVersionCheckerFails() {
    // Given:
    doThrow(new Exception("FOO")).when(ksqlVersionChecker).start();

    // When:
    ksqlVersionCheckerAgent.start(KsqlModuleType.SERVER, properties);
  }