@Override
  public Set<MaintenanceEvent> readEvents(Duration timeout) throws SamplingException {
    LOG.debug("Reading maintenance events.");
    long eventReadPeriodEndMs = _kafkaCruiseControl.timeMs();
    if (refreshPartitionAssignment()) {
      _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs;
      return Collections.emptySet();
    }

    long timeoutEndMs = eventReadPeriodEndMs + timeout.toMillis();
    Set<MaintenanceEvent> maintenanceEvents = new HashSet<>();
    try {
      Map<TopicPartition, Long> endOffsets = seekToRelevantOffsets();
      LOG.debug("Started to consume from maintenance event topic partitions {}.", _currentPartitionAssignment);
      _consumer.resume(_consumer.paused());
      Set<TopicPartition> partitionsToPause = new HashSet<>();

      do {
        ConsumerRecords<String, MaintenancePlan> records = _consumer.poll(timeout);
        for (ConsumerRecord<String, MaintenancePlan> record : records) {
          if (record == null) {
            // This means that the record cannot be parsed because the maintenance plan version is not supported. It might
            // happen when existing maintenance plans have been updated and the current code is still old. We simply ignore
            // that plan in this case (see MaintenancePlanSerde.MaintenancePlanTypeAdapter#verifyTypeAndVersion(String, byte).
            LOG.warn("Cannot parse record, please update your Cruise Control version.");
            continue;
          }

          long planGenerationTimeMs = record.value().timeMs();
          if (planGenerationTimeMs + _maintenancePlanExpirationMs < eventReadPeriodEndMs) {
            LOG.warn("Discarding the expired plan {}. (Expired: {} Evaluated: {}).", record.value(),
                     planGenerationTimeMs + _maintenancePlanExpirationMs, eventReadPeriodEndMs);
          } else if (planGenerationTimeMs >= eventReadPeriodEndMs) {
            TopicPartition tp = new TopicPartition(record.topic(), record.partition());
            LOG.debug("Saw plan {} generated after the end time of event read period {}. Pausing {} at offset {}.",
                      record.value(), eventReadPeriodEndMs, tp, record.offset());
            partitionsToPause.add(tp);
          } else {
            addMaintenancePlan(record.value(), maintenanceEvents);
          }
        }

        if (!partitionsToPause.isEmpty()) {
          _consumer.pause(partitionsToPause);
          partitionsToPause.clear();
        }
      } while (!consumptionDone(_consumer, endOffsets) && _kafkaCruiseControl.timeMs() < timeoutEndMs);

      if (maintenanceEvents.size() > 0) {
        LOG.info("Retrieved {} maintenance plans from partitions {} (range [{},{}]).", maintenanceEvents.size(),
                 _currentPartitionAssignment, _lastEventReadPeriodEndTimeMs, eventReadPeriodEndMs);
      }
    } finally {
      _lastEventReadPeriodEndTimeMs = eventReadPeriodEndMs;
    }

    return maintenanceEvents;
  }
@Test
  public void testMaintenanceEventTopicCreationUpdateAndRead()
      throws ExecutionException, InterruptedException, SamplingException {
    // Verify that the maintenance event topic has been created with the desired properties.
    verify(TEST_TOPIC_PARTITION_COUNT, TEST_TOPIC_REPLICATION_FACTOR, TEST_TOPIC_RETENTION_TIME_MS, false);

    // Verify that the maintenance event topic properties can be updated if the topic already exists with different properties.
    String newPartitionCount = String.valueOf(Integer.parseInt(TEST_TOPIC_PARTITION_COUNT) * 2);
    String newRF = String.valueOf(Short.parseShort(TEST_TOPIC_REPLICATION_FACTOR) + 1);
    String newRetentionMs = String.valueOf(Long.MAX_VALUE);

    KafkaCruiseControl mockKafkaCruiseControl = EasyMock.mock(KafkaCruiseControl.class);
    Map<String, Object> parameterConfigOverrides = Map.of(KAFKA_CRUISE_CONTROL_OBJECT_CONFIG, mockKafkaCruiseControl,
                                                          MAINTENANCE_EVENT_TOPIC_REPLICATION_FACTOR_CONFIG, newRF,
                                                          MAINTENANCE_EVENT_TOPIC_PARTITION_COUNT_CONFIG, newPartitionCount,
                                                          MAINTENANCE_EVENT_TOPIC_RETENTION_MS_CONFIG, newRetentionMs);

    // The current time is expected to cause (1) a valid rebalance plan creation, but (2) an expired demote broker plan.
    long currentMockTime = TEST_REBALANCE_PLAN_TIME + DEFAULT_MAINTENANCE_PLAN_EXPIRATION_MS;
    EasyMock.expect(mockKafkaCruiseControl.timeMs()).andReturn(currentMockTime).anyTimes();
    EasyMock.expect(mockKafkaCruiseControl.adminClient())
            .andReturn(createAdminClient(KafkaCruiseControlUtils.parseAdminClientConfigs(_config))).anyTimes();
    EasyMock.expect(mockKafkaCruiseControl.config()).andReturn(_config).anyTimes();

    EasyMock.replay(mockKafkaCruiseControl);

    MaintenanceEventReader maintenanceEventReader = _config.getConfiguredInstance(AnomalyDetectorConfig.MAINTENANCE_EVENT_READER_CLASS_CONFIG,
                                                                                  MaintenanceEventTopicReader.class,
                                                                                  parameterConfigOverrides);

    assertNotNull(maintenanceEventReader);
    verify(newPartitionCount, newRF, newRetentionMs, true);
    Set<MaintenanceEvent> events = maintenanceEventReader.readEvents(TEST_TIMEOUT);
    EasyMock.verify(mockKafkaCruiseControl);
    assertEquals(1, events.size());
    MaintenanceEvent maintenanceEvent = events.iterator().next();
    assertEquals(MAINTENANCE_EVENT, maintenanceEvent.anomalyType());
    assertEquals(currentMockTime, maintenanceEvent.detectionTimeMs());
    assertEquals(REBALANCE, maintenanceEvent.maintenanceEventType());
  }