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.bookkeeper.meta;
  
  import static java.nio.charset.StandardCharsets.UTF_8;
  
  import com.google.common.collect.Lists;
  import java.io.IOException;
  import java.util.Base64;
  import java.util.Optional;
  import java.util.Random;
  import org.apache.bookkeeper.client.LedgerMetadataBuilder;
  import org.apache.bookkeeper.client.api.DigestType;
  import org.apache.bookkeeper.client.api.LedgerMetadata;
  import org.apache.bookkeeper.net.BookieSocketAddress;
  import org.junit.Assert;
  import org.junit.Test;
  
  /**
   * Test Ledger Metadata serialization and deserialization.
   */
  public class TestLedgerMetadataSerDe {
      // as used in 4.0.x & 4.1.x
      private static final String version1 =
          "Qm9va2llTWV0YWRhdGFGb3JtYXRWZXJzaW9uCTEKMgozCjAKMAkxOTIuMC4yLjE6MTIzNAkxOTIu"
          + "MC4yLjI6MTIzNAkxOTIuMC4yLjM6MTIzNAotMTAyCUNMT1NFRA==";
  
      // as used in 4.2.x & 4.3.x (text protobuf based metadata, password and digest introduced)
      private static final String version2 =
          "Qm9va2llTWV0YWRhdGFGb3JtYXRWZXJzaW9uCTIKcXVvcnVtU2l6ZTogMgplbnNlbWJsZVNpemU6I"
          + "DMKbGVuZ3RoOiAwCmxhc3RFbnRyeUlkOiAtMQpzdGF0ZTogSU5fUkVDT1ZFUlkKc2VnbWVudCB7"
          + "CiAgZW5zZW1ibGVNZW1iZXI6ICIxOTIuMC4yLjE6MTIzNCIKICBlbnNlbWJsZU1lbWJlcjogIjE"
          + "5Mi4wLjIuMjoxMjM0IgogIGVuc2VtYmxlTWVtYmVyOiAiMTkyLjAuMi4zOjEyMzQiCiAgZmlyc3"
          + "RFbnRyeUlkOiAwCn0KZGlnZXN0VHlwZTogQ1JDMzIKcGFzc3dvcmQ6ICJwYXNzd2QiCmFja1F1b"
          + "3J1bVNpemU6IDIK";
  
      // version 2 + ctime, as used in 4.4.x to 4.8.x (ctime is optional from 4.6.x onwards)
      private static final String version2ctime =
          "Qm9va2llTWV0YWRhdGFGb3JtYXRWZXJzaW9uCTIKcXVvcnVtU2l6ZTogMgplbnNlbWJsZVNpemU6I"
          + "DMKbGVuZ3RoOiAwCmxhc3RFbnRyeUlkOiAtMQpzdGF0ZTogSU5fUkVDT1ZFUlkKc2VnbWVudCB7"
          + "CiAgZW5zZW1ibGVNZW1iZXI6ICIxOTIuMC4yLjE6MTIzNCIKICBlbnNlbWJsZU1lbWJlcjogIjE"
          + "5Mi4wLjIuMjoxMjM0IgogIGVuc2VtYmxlTWVtYmVyOiAiMTkyLjAuMi4zOjEyMzQiCiAgZmlyc3"
          + "RFbnRyeUlkOiAwCn0KZGlnZXN0VHlwZTogQ1JDMzIKcGFzc3dvcmQ6ICJwYXNzd2QiCmFja1F1b"
          + "3J1bVNpemU6IDIKY3RpbWU6IDE1NDQwMDIzODMwNzUK";
  
      // version 3, since 4.9.x, protobuf binary format
      private static final String version3 =
          "Qm9va2llTWV0YWRhdGFGb3JtYXRWZXJzaW9uCTMKYAgCEAMYACD///////////8BKAEyMgoOMTkyL"
          + "jAuMi4xOjMxODEKDjE5Mi4wLjIuMjozMTgxCg4xOTIuMC4yLjM6MzE4MRAAOANCBmZvb2JhckgB"
          + "UP///////////wFgAA==";
  
      private static void testDecodeEncode(String encoded) throws Exception {
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          LedgerMetadata md = serDe.parseConfig(Base64.getDecoder().decode(encoded), 59L, Optional.empty());
          String reserialized = Base64.getEncoder().encodeToString(serDe.serialize(md));
  
          Assert.assertEquals(encoded, reserialized);
      }
  
      @Test
      public void testVersion1SerDe() throws Exception {
          testDecodeEncode(version1);
      }
  
      @Test
      public void testVersion2SerDe() throws Exception {
          testDecodeEncode(version2);
      }
  
      @Test
      public void testVersion2CtimeSerDe() throws Exception {
          testDecodeEncode(version2ctime);
      }
  
      @Test
      public void testVersion3SerDe() throws Exception {
          testDecodeEncode(version3);
      }
  
      @Test(expected = IOException.class)
      public void testJunkSerDe() throws Exception {
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          String junk = "";
          serDe.parseConfig(junk.getBytes(UTF_8), 59L, Optional.empty());
      }
  
      @Test(expected = IOException.class)
      public void testJunk2SerDe() throws Exception {
          byte[] randomBytes = new byte[1000];
          new Random().nextBytes(randomBytes);
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          serDe.parseConfig(randomBytes, 59L, Optional.empty());
      }
  
      @Test(expected = IOException.class)
      public void testJunkVersionSerDe() throws Exception {
          byte[] junkVersion = "BookieMetadataFormatVersion\tfoobar\nblahblah".getBytes(UTF_8);
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          serDe.parseConfig(junkVersion, 59L, Optional.empty());
      }
  
      @Test(expected = IOException.class)
      public void testVeryLongVersionSerDe() throws Exception {
          byte[] veryLongVersion = "BookieMetadataFormatVersion\t123456789123456789\nblahblah".getBytes(UTF_8);
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          serDe.parseConfig(veryLongVersion, 59L, Optional.empty());
      }
  
      @Test
      public void testPeggedToV3SerDe() throws Exception {
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          LedgerMetadata metadata = LedgerMetadataBuilder.create().withId(13L)
              .withEnsembleSize(3).withWriteQuorumSize(2).withAckQuorumSize(1)
              .withPassword("foobar".getBytes(UTF_8)).withDigestType(DigestType.CRC32C)
              .newEnsembleEntry(0L, Lists.newArrayList(new BookieSocketAddress("192.0.2.1", 3181).toBookieId(),
                                                       new BookieSocketAddress("192.0.2.2", 3181).toBookieId(),
                                                       new BookieSocketAddress("192.0.2.3", 3181).toBookieId()))
              .build();
          byte[] encoded = serDe.serialize(metadata);
  
          LedgerMetadata decoded = serDe.parseConfig(encoded, 59L, Optional.empty());
          Assert.assertEquals(LedgerMetadataSerDe.METADATA_FORMAT_VERSION_3, decoded.getMetadataFormatVersion());
      }
  
      @Test
      public void testStoreSystemtimeAsLedgerCtimeEnabledWithNewerVersion()
              throws Exception {
          LedgerMetadata lm = LedgerMetadataBuilder.create().withId(13L)
              .withEnsembleSize(3).withWriteQuorumSize(2).withAckQuorumSize(1)
              .withPassword("foobar".getBytes(UTF_8)).withDigestType(DigestType.CRC32C)
              .newEnsembleEntry(0L, Lists.newArrayList(new BookieSocketAddress("192.0.2.1", 1234).toBookieId(),
                                        new BookieSocketAddress("192.0.2.2", 1234).toBookieId(),
                                        new BookieSocketAddress("192.0.2.3", 1234).toBookieId()))
              .withCreationTime(123456L)
              .storingCreationTime(true)
              .build();
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          byte[] serialized = serDe.serialize(lm);
          LedgerMetadata deserialized = serDe.parseConfig(serialized, 59L, Optional.of(654321L));
          Assert.assertEquals(deserialized.getCtime(), 123456L);
  
          // give it another round
          LedgerMetadata deserialized2 = serDe.parseConfig(serDe.serialize(deserialized), 59L, Optional.of(98765L));
          Assert.assertEquals(deserialized2.getCtime(), 123456L);
      }
  
      @Test
      public void testStoreSystemtimeAsLedgerCtimeDisabledWithNewerVersion()
              throws Exception {
          LedgerMetadata lm = LedgerMetadataBuilder.create().withId(13L)
              .withEnsembleSize(3).withWriteQuorumSize(2).withAckQuorumSize(1)
              .withPassword("foobar".getBytes(UTF_8)).withDigestType(DigestType.CRC32C)
              .newEnsembleEntry(0L, Lists.newArrayList(new BookieSocketAddress("192.0.2.1", 1234).toBookieId(),
                                        new BookieSocketAddress("192.0.2.2", 1234).toBookieId(),
                                        new BookieSocketAddress("192.0.2.3", 1234).toBookieId()))
              .build();
          LedgerMetadataSerDe serDe = new LedgerMetadataSerDe();
          byte[] serialized = serDe.serialize(lm);
          LedgerMetadata deserialized = serDe.parseConfig(serialized, 59L, Optional.of(654321L));
          Assert.assertEquals(654321L, deserialized.getCtime());
  
          // give it another round
          LedgerMetadata deserialized2 = serDe.parseConfig(serDe.serialize(deserialized), 59L, Optional.of(98765L));
          Assert.assertEquals(98765L, deserialized2.getCtime());
      }
  }
  