/*
 * 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.accumulo.test.functional;

import static org.junit.Assert.assertTrue;

import java.util.Base64;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.BatchWriter;
import org.apache.accumulo.core.client.BatchWriterConfig;
import org.apache.accumulo.core.conf.Property;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.harness.AccumuloClusterHarness;
import org.apache.accumulo.miniclusterImpl.MiniAccumuloConfigImpl;
import org.apache.accumulo.server.ServerContext;
import org.apache.accumulo.server.log.WalStateManager.WalState;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.RawLocalFileSystem;
import org.apache.hadoop.io.Text;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ManyWriteAheadLogsIT extends AccumuloClusterHarness {

  private static final Logger log = LoggerFactory.getLogger(ManyWriteAheadLogsIT.class);

  @Override
  public void configureMiniCluster(MiniAccumuloConfigImpl cfg, Configuration hadoopCoreSite) {
    // configure a smaller walog size so the walogs will roll frequently in the test
    cfg.setProperty(Property.TSERV_WALOG_MAX_SIZE, "1M");
    cfg.setProperty(Property.GC_CYCLE_DELAY, "1");
    cfg.setProperty(Property.GC_CYCLE_START, "1");
    cfg.setProperty(Property.MASTER_RECOVERY_DELAY, "1s");
    cfg.setProperty(Property.TSERV_MAJC_DELAY, "1");
    cfg.setProperty(Property.INSTANCE_ZK_TIMEOUT, "15s");
    // idle compactions may addess the problem this test is creating, however they will not prevent
    // lots of closed WALs for all write patterns. This test ensures code that directly handles many
    // tablets referencing many different WALs is working.
    cfg.setProperty(Property.TABLE_MINC_COMPACT_IDLETIME, "1h");
    cfg.setNumTservers(1);
    hadoopCoreSite.set("fs.file.impl", RawLocalFileSystem.class.getName());
  }

  @Override
  protected int defaultTimeoutSeconds() {
    return 10 * 60;
  }

  /**
   * This creates a situation where many tablets reference many different write ahead logs. However
   * not single tablet references a lot of write ahead logs. Want to ensure the tablet server forces
   * minor compactions for this situation.
   */
  @Test
  public void testMany() throws Exception {
    SortedSet<Text> splits = new TreeSet<>();
    for (int i = 1; i < 100; i++) {
      splits.add(new Text(String.format("%05x", i * 100)));
    }

    ServerContext ctx = getServerContext();

    try (AccumuloClient c = createAccumuloClient()) {
      String[] tableNames = getUniqueNames(2);

      String manyWALsTable = tableNames[0];
      String rollWALsTable = tableNames[1];
      c.tableOperations().create(manyWALsTable);
      c.tableOperations().addSplits(manyWALsTable, splits);

      c.tableOperations().create(rollWALsTable);

      Random rand = new Random();

      Set<String> allWalsSeen = new HashSet<>();

      addOpenWals(ctx, allWalsSeen);

      // This test creates the table manyWALsTable with a lot of tablets and writes a little bit to
      // each tablet. In between writing a little bit to each tablet a lot of data is written to
      // another table called rollWALsTable. Writing a lot causes the write ahead logs to roll. This
      // write pattern should cause the tablets in the manyWALsTable table to reference many closed
      // WALs. If nothing is done about all of these closed WALs, then it could cause a large burden
      // at recovery time.

      try (BatchWriter manyWALsWriter = c.createBatchWriter(manyWALsTable, new BatchWriterConfig());
          BatchWriter rollWALsWriter = c.createBatchWriter(rollWALsTable,
              new BatchWriterConfig())) {

        byte[] val = new byte[768];

        for (int i = 0; i < 100; i++) {
          int startRow = i * 100;

          // write a small amount of data to each tablet in the table
          for (int j = 0; j < 10; j++) {
            int row = startRow + j;
            Mutation m = new Mutation(String.format("%05x", row));
            rand.nextBytes(val);
            m.put("f", "q", "v");

            manyWALsWriter.addMutation(m);
          }
          manyWALsWriter.flush();

          // write a lot of data to second table to forces the logs to roll
          for (int j = 0; j < 1000; j++) {
            Mutation m = new Mutation(String.format("%03d", j));
            rand.nextBytes(val);

            m.put("f", "q", Base64.getEncoder().encodeToString(val));

            rollWALsWriter.addMutation(m);
          }

          rollWALsWriter.flush();

          // keep track of the open WALs as the test runs. Should see a lot of open WALs over the
          // lifetime of the test, but never a lot at any one time.
          addOpenWals(ctx, allWalsSeen);
        }
      }

      assertTrue("Number of WALs seen was less than expected " + allWalsSeen.size(),
          allWalsSeen.size() >= 50);

      // the total number of closed write ahead logs should get small
      int closedLogs = countClosedWals(ctx);
      while (closedLogs > 3) {
        log.debug("Waiting for wals to shrink " + closedLogs);
        Thread.sleep(250);
        closedLogs = countClosedWals(ctx);
      }
    }
  }

  private void addOpenWals(ServerContext c, Set<String> allWalsSeen) throws Exception {
    Map<String,WalState> wals = WALSunnyDayIT._getWals(c);
    Set<Entry<String,WalState>> es = wals.entrySet();
    int open = 0;
    for (Entry<String,WalState> entry : es) {
      if (entry.getValue() == WalState.OPEN) {
        open++;
        allWalsSeen.add(entry.getKey());
      }
    }

    assertTrue("Open WALs not in expected range " + open, open > 0 && open < 4);
  }

  private int countClosedWals(ServerContext c) throws Exception {
    int count = 0;
    Map<String,WalState> wals = WALSunnyDayIT._getWals(c);
    for (WalState ws : wals.values()) {
      if (ws == WalState.CLOSED) {
        count++;
      }
    }

    return count;
  }
}

@@ -886,16 +886,51 @@
 tTrue;%0A%0A
+import java.security.SecureRandom;%0A
 import j
@@ -3898,16 +3898,22 @@
 d = new 
+Secure
 Random()

Use secureRandom in ManyWriteAheadLogsIT