content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.kafka.metadata;
  
  import org.apache.kafka.common.Endpoint;
  import org.apache.kafka.common.Node;
  import org.apache.kafka.common.Uuid;
  import org.apache.kafka.common.metadata.RegisterControllerRecord;
  import org.apache.kafka.common.security.auth.SecurityProtocol;
  import org.apache.kafka.image.writer.ImageWriterOptions;
  import org.apache.kafka.server.common.ApiMessageAndVersion;
  import org.apache.kafka.server.common.MetadataVersion;
  
  import org.junit.jupiter.api.Test;
  import org.junit.jupiter.api.Timeout;
  
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertNotEquals;
  
  
  @Timeout(value = 40)
  public class ControllerRegistrationTest {
      static <K, V> Map<K, V> doubleMap(K k1, V v1, K k2, V v2) {
          LinkedHashMap<K, V> map = new LinkedHashMap<>();
          map.put(k1, v1);
          map.put(k2, v2);
          return map;
      }
  
      private static final List<ControllerRegistration> REGISTRATIONS = List.of(
          new ControllerRegistration.Builder().
              setId(0).
              setIncarnationId(Uuid.fromString("ycRmGrOFQru7HXf6fOybZQ")).
              setZkMigrationReady(true).
              setListeners(doubleMap(
                  "PLAINTEXT", new Endpoint("PLAINTEXT", SecurityProtocol.PLAINTEXT, "localhost", 9107),
                  "SSL", new Endpoint("SSL", SecurityProtocol.SSL, "localhost", 9207))).
              setSupportedFeatures(Map.of(MetadataVersion.FEATURE_NAME, VersionRange.of(1, 10))).
              build(),
          new ControllerRegistration.Builder().
              setId(1).
              setIncarnationId(Uuid.fromString("ubT_wuD6R3uopZ_lV76dQg")).
              setZkMigrationReady(true).
              setListeners(doubleMap(
                  "PLAINTEXT", new Endpoint("PLAINTEXT", SecurityProtocol.PLAINTEXT, "localhost", 9108),
                  "SSL", new Endpoint("SSL", SecurityProtocol.SSL, "localhost", 9208))).
              setSupportedFeatures(Map.of(MetadataVersion.FEATURE_NAME, VersionRange.of(1, 10))).
              build(),
          new ControllerRegistration.Builder().
              setId(2).
              setIncarnationId(Uuid.fromString("muQS341gRIeNh9Ps7reDSw")).
              setZkMigrationReady(false).
              setListeners(doubleMap(
                  "PLAINTEXT", new Endpoint("PLAINTEXT", SecurityProtocol.PLAINTEXT, "localhost", 9109),
                  "SSL", new Endpoint("SSL", SecurityProtocol.SSL, "localhost", 9209))).
              setSupportedFeatures(Map.of(MetadataVersion.FEATURE_NAME, VersionRange.of(1, 10))).
              build()
      );
  
      @Test
      public void testValues() {
          assertEquals(0, REGISTRATIONS.get(0).id());
          assertEquals(1, REGISTRATIONS.get(1).id());
          assertEquals(2, REGISTRATIONS.get(2).id());
      }
  
      @Test
      public void testEquals() {
          assertNotEquals(REGISTRATIONS.get(0), REGISTRATIONS.get(1));
          assertNotEquals(REGISTRATIONS.get(1), REGISTRATIONS.get(0));
          assertNotEquals(REGISTRATIONS.get(0), REGISTRATIONS.get(2));
          assertNotEquals(REGISTRATIONS.get(2), REGISTRATIONS.get(0));
          assertEquals(REGISTRATIONS.get(0), REGISTRATIONS.get(0));
          assertEquals(REGISTRATIONS.get(1), REGISTRATIONS.get(1));
          assertEquals(REGISTRATIONS.get(2), REGISTRATIONS.get(2));
      }
  
      @Test
      public void testToString() {
          assertEquals("ControllerRegistration(id=1, " +
              "incarnationId=ubT_wuD6R3uopZ_lV76dQg, " +
              "zkMigrationReady=true, " +
              "listeners=[" +
              "Endpoint(listenerName='PLAINTEXT', securityProtocol=PLAINTEXT, host='localhost', port=9108), " +
              "Endpoint(listenerName='SSL', securityProtocol=SSL, host='localhost', port=9208)]" +
              ", supportedFeatures={metadata.version: 1-10})",
              REGISTRATIONS.get(1).toString());
      }
  
      @Test
      public void testFromRecordAndToRecord() {
          testRoundTrip(REGISTRATIONS.get(0));
          testRoundTrip(REGISTRATIONS.get(1));
          testRoundTrip(REGISTRATIONS.get(2));
      }
  
      private void testRoundTrip(ControllerRegistration registration) {
          ApiMessageAndVersion messageAndVersion = registration.
              toRecord(new ImageWriterOptions.Builder(MetadataVersion.latestProduction()).build());
          ControllerRegistration registration2 = new ControllerRegistration.Builder(
              (RegisterControllerRecord) messageAndVersion.message()).build();
          assertEquals(registration, registration2);
          ApiMessageAndVersion messageAndVersion2 = registration2.
              toRecord(new ImageWriterOptions.Builder(MetadataVersion.latestProduction()).build());
          assertEquals(messageAndVersion, messageAndVersion2);
      }
  
      @Test
      public void testToNode() {
          assertEquals(Optional.empty(), REGISTRATIONS.get(0).node("NONEXISTENT"));
          assertEquals(Optional.of(new Node(0, "localhost", 9107, null)),
              REGISTRATIONS.get(0).node("PLAINTEXT"));
          assertEquals(Optional.of(new Node(0, "localhost", 9207, null)),
              REGISTRATIONS.get(0).node("SSL"));
      }
  }
  