public Map<String, Object> populateAnomalyDetails() {
    // Goal violation has one more field than other anomaly types.
    Map<String, Object> anomalyDetails = new HashMap<>();
    anomalyDetails.put(_isJson ? DETECTION_MS : DETECTION_DATE,
                       _isJson ? _anomalyState.detectionMs() : utcDateFor(_anomalyState.detectionMs()));
    anomalyDetails.put(STATUS, _anomalyState.status());
    anomalyDetails.put(ANOMALY_ID, _anomalyState.anomalyId());
    anomalyDetails.put(_isJson ? STATUS_UPDATE_MS : STATUS_UPDATE_DATE,
                       _isJson ? _anomalyState.statusUpdateMs() : utcDateFor(_anomalyState.statusUpdateMs()));
    switch ((KafkaAnomalyType) _anomalyType) {
      case GOAL_VIOLATION:
        GoalViolations goalViolations = (GoalViolations) _anomalyState.anomaly();
        Map<Boolean, List<String>> violatedGoalsByFixability = goalViolations.violatedGoalsByFixability();
        anomalyDetails.put(FIXABLE_VIOLATED_GOALS, violatedGoalsByFixability.getOrDefault(true, Collections.emptyList()));
        anomalyDetails.put(UNFIXABLE_VIOLATED_GOALS, violatedGoalsByFixability.getOrDefault(false, Collections.emptyList()));
        if (_hasFixStarted) {
          anomalyDetails.put(OPTIMIZATION_RESULT, goalViolations.optimizationResult(_isJson));
        }
        break;
      case BROKER_FAILURE:
        BrokerFailures brokerFailures = (BrokerFailures) _anomalyState.anomaly();
        anomalyDetails.put(FAILED_BROKERS_BY_TIME_MS, brokerFailures.failedBrokers());
        if (_hasFixStarted) {
          anomalyDetails.put(OPTIMIZATION_RESULT, brokerFailures.optimizationResult(_isJson));
        }
        break;
      case DISK_FAILURE:
        DiskFailures diskFailures = (DiskFailures) _anomalyState.anomaly();
        anomalyDetails.put(FAILED_DISKS_BY_TIME_MS, diskFailures.failedDisks());
        if (_hasFixStarted) {
          anomalyDetails.put(OPTIMIZATION_RESULT, diskFailures.optimizationResult(_isJson));
        }
        break;
      case METRIC_ANOMALY:
        KafkaMetricAnomaly metricAnomaly = (KafkaMetricAnomaly) _anomalyState.anomaly();
        anomalyDetails.put(DESCRIPTION, metricAnomaly.description());
        if (_hasFixStarted) {
          anomalyDetails.put(OPTIMIZATION_RESULT, metricAnomaly.optimizationResult(_isJson));
        }
        break;
      case TOPIC_ANOMALY:
        TopicAnomaly topicAnomaly = (TopicAnomaly) _anomalyState.anomaly();
        anomalyDetails.put(DESCRIPTION, topicAnomaly.toString());
        if (_hasFixStarted) {
          anomalyDetails.put(OPTIMIZATION_RESULT, topicAnomaly.optimizationResult(_isJson));
        }
        break;
      case MAINTENANCE_EVENT:
        MaintenanceEvent maintenanceEvent = (MaintenanceEvent) _anomalyState.anomaly();
        anomalyDetails.put(DESCRIPTION, maintenanceEvent.toString());
        if (_hasFixStarted) {
          anomalyDetails.put(OPTIMIZATION_RESULT, maintenanceEvent.optimizationResult(_isJson));
        }
        break;
      default:
        throw new IllegalStateException("Unrecognized anomaly type " + _anomalyType);
    }
    return anomalyDetails;
  }
@Test
  public void testPopulateAnomalyDetails() {
    AnomalyState mockAnomalyState = EasyMock.mock(AnomalyState.class);

    for (KafkaAnomalyType anomalyType : KafkaAnomalyType.cachedValues()) {
      AnomalyDetails details = new AnomalyDetails(mockAnomalyState, anomalyType, false, false);
      EasyMock.expect(mockAnomalyState.detectionMs()).andReturn(MOCK_DETECTION_MS).once();
      EasyMock.expect(mockAnomalyState.status()).andReturn(DETECTED).once();
      EasyMock.expect(mockAnomalyState.anomalyId()).andReturn(MOCK_ANOMALY_ID).once();
      EasyMock.expect(mockAnomalyState.statusUpdateMs()).andReturn(MOCK_STATUS_UPDATE_MS).once();

      switch (anomalyType) {
        case GOAL_VIOLATION:
          GoalViolations goalViolations = EasyMock.mock(GoalViolations.class);
          EasyMock.expect(goalViolations.violatedGoalsByFixability()).andReturn(VIOLATED_GOALS_BY_FIXABILITY).once();
          replayAndVerify(mockAnomalyState, details, goalViolations);
          break;
        case BROKER_FAILURE:
          BrokerFailures brokerFailures = EasyMock.mock(BrokerFailures.class);
          EasyMock.expect(brokerFailures.failedBrokers()).andReturn(FAILED_BROKERS).once();
          replayAndVerify(mockAnomalyState, details, brokerFailures);
          break;
        case DISK_FAILURE:
          DiskFailures diskFailures = EasyMock.mock(DiskFailures.class);
          EasyMock.expect(diskFailures.failedDisks()).andReturn(FAILED_DISKS).once();
          replayAndVerify(mockAnomalyState, details, diskFailures);
          break;
        case METRIC_ANOMALY:
          KafkaMetricAnomaly kafkaMetricAnomaly = EasyMock.mock(KafkaMetricAnomaly.class);
          EasyMock.expect(kafkaMetricAnomaly.description()).andReturn(MOCK_DESCRIPTION).once();
          replayAndVerify(mockAnomalyState, details, kafkaMetricAnomaly);
          break;
        case TOPIC_ANOMALY:
          TopicAnomaly topicAnomaly = EasyMock.mock(TopicAnomaly.class);
          // Note: EasyMock provides a built-in behavior for equals(), toString(), hashCode() and finalize() even for class mocking.
          // Hence, we cannot record our own behavior for topicAnomaly.toString().
          replayAndVerify(mockAnomalyState, details, topicAnomaly);
          break;
        case MAINTENANCE_EVENT:
          MaintenanceEvent maintenanceEvent = EasyMock.mock(MaintenanceEvent.class);
          // Note: EasyMock provides a built-in behavior for equals(), toString(), hashCode() and finalize() even for class mocking.
          // Hence, we cannot record our own behavior for maintenanceEvent.toString().
          replayAndVerify(mockAnomalyState, details, maintenanceEvent);
          break;
        default:
          throw new IllegalStateException("Unrecognized anomaly type " + anomalyType);
      }
    }
  }