// Path: logging/src/main/java/org/kitesdk/examples/logging/CreateDataset.java
// public class CreateDataset extends Configured implements Tool {
// 
//   @Override
//   public int run(String[] args) throws Exception {
// 
//     // Create a dataset of events with the Avro schema
//     DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
//         .schemaUri("resource:event.avsc")
//         .build();
//     Datasets.create("dataset:hive:/tmp/data/default/events", descriptor);
// 
//     return 0;
//   }
// 
//   public static void main(String... args) throws Exception {
//     int rc = ToolRunner.run(new CreateDataset(), args);
//     System.exit(rc);
//   }
// }
// 
// Path: logging/src/main/java/org/kitesdk/examples/logging/DeleteDataset.java
// public class DeleteDataset extends Configured implements Tool {
// 
//   @Override
//   public int run(String[] args) throws Exception {
// 
//     // Drop the events dataset
//     boolean success = Datasets.delete("dataset:hive:/tmp/data/default/events");
// 
//     return success ? 0 : 1;
//   }
// 
//   public static void main(String... args) throws Exception {
//     int rc = ToolRunner.run(new DeleteDataset(), args);
//     System.exit(rc);
//   }
// }
// 
// Path: logging/src/main/java/org/kitesdk/examples/logging/ReadDataset.java
// public class ReadDataset extends Configured implements Tool {
// 
//   @Override
//   public int run(String[] args) throws Exception {
// 
//     // Load the events dataset
//     Dataset<GenericRecord> events = Datasets.load("dataset:hive:/tmp/data/default/events");
// 
//     // Get a reader for the dataset and read all the events
//     DatasetReader<GenericRecord> reader = events.newReader();
//     try {
//       for (GenericRecord event : reader) {
//         System.out.println(event);
//       }
//     } finally {
//       reader.close();
//     }
// 
//     return 0;
//   }
// 
//   public static void main(String... args) throws Exception {
//     int rc = ToolRunner.run(new ReadDataset(), args);
//     System.exit(rc);
//   }
// }


import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.io.PrintStream;
import org.apache.catalina.core.AprLifecycleListener;
import org.apache.catalina.core.StandardServer;
import org.apache.catalina.startup.Tomcat;
import org.apache.flume.clients.log4jappender.Log4jAppender;
import org.apache.hadoop.util.Tool;
import org.apache.http.client.fluent.Request;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.hamcrest.Matcher;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.kitesdk.examples.logging.CreateDataset;
import org.kitesdk.examples.logging.DeleteDataset;
import org.kitesdk.examples.logging.ReadDataset;
import org.kitesdk.minicluster.FlumeService;
import org.kitesdk.minicluster.HdfsService;
import org.kitesdk.minicluster.HiveService;
import org.kitesdk.minicluster.MiniCluster;
import static org.hamcrest.CoreMatchers.any;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.matchers.JUnitMatchers.containsString;
  public static void stopCluster() {
    try {
      cluster.stop();
    } catch (Exception e) {
      // ignore problems during shutdown
    }
  }

  private static void configureLog4j() throws Exception {
    // configuration is done programmatically and not in log4j.properties so that so we
    // can defer initialization to after the Flume Avro RPC source port is running
    Log4jAppender appender = new Log4jAppender();
    appender.setName("flume");
    appender.setHostname("localhost");
    appender.setPort(41415);
    appender.setUnsafeMode(true);
    appender.activateOptions();

    Logger.getLogger("org.kitesdk.examples.logging.webapp.LoggingServlet").addAppender(appender);
    Logger.getLogger("org.kitesdk.examples.logging.webapp.LoggingServlet").setLevel(Level.INFO);
  }

  @Test
  public void test() throws Exception {
    run(new CreateDataset());
    for (int i = 1; i <= 10; i++) {
      get("http://localhost:8080/logging-webapp/send?message=Hello-" + i);
    }
    Thread.sleep(40000); // wait for events to be flushed to HDFS
    run(containsString("{\"id\": 10, \"message\": \"Hello-10\"}"), new ReadDataset());
// Path: logging/src/main/java/org/kitesdk/examples/logging/CreateDataset.java
// public class CreateDataset extends Configured implements Tool {
// 
//   @Override
//   public int run(String[] args) throws Exception {
// 
//     // Create a dataset of events with the Avro schema
//     DatasetDescriptor descriptor = new DatasetDescriptor.Builder()
//         .schemaUri("resource:event.avsc")
//         .build();
//     Datasets.create("dataset:hive:/tmp/data/default/events", descriptor);
// 
//     return 0;
//   }
// 
//   public static void main(String... args) throws Exception {
//     int rc = ToolRunner.run(new CreateDataset(), args);
//     System.exit(rc);
//   }
// }
// 
// Path: logging/src/main/java/org/kitesdk/examples/logging/DeleteDataset.java
// public class DeleteDataset extends Configured implements Tool {
// 
//   @Override
//   public int run(String[] args) throws Exception {
// 
//     // Drop the events dataset
//     boolean success = Datasets.delete("dataset:hive:/tmp/data/default/events");
// 
//     return success ? 0 : 1;
//   }
// 
//   public static void main(String... args) throws Exception {
//     int rc = ToolRunner.run(new DeleteDataset(), args);
//     System.exit(rc);
//   }
// }
// 
// Path: logging/src/main/java/org/kitesdk/examples/logging/ReadDataset.java
// public class ReadDataset extends Configured implements Tool {
// 
//   @Override
//   public int run(String[] args) throws Exception {
// 
//     // Load the events dataset
//     Dataset<GenericRecord> events = Datasets.load("dataset:hive:/tmp/data/default/events");
// 
//     // Get a reader for the dataset and read all the events
//     DatasetReader<GenericRecord> reader = events.newReader();
//     try {
//       for (GenericRecord event : reader) {
//         System.out.println(event);
//       }
//     } finally {
//       reader.close();
//     }
// 
//     return 0;
//   }
// 
//   public static void main(String... args) throws Exception {
//     int rc = ToolRunner.run(new ReadDataset(), args);
//     System.exit(rc);
//   }
// }


// Path: logging-webapp/src/test/java/org/kitesdk/examples/logging/webapp/ITLoggingWebapp.java
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.io.PrintStream;
import org.apache.catalina.core.AprLifecycleListener;
import org.apache.catalina.core.StandardServer;
import org.apache.catalina.startup.Tomcat;
import org.apache.flume.clients.log4jappender.Log4jAppender;
import org.apache.hadoop.util.Tool;
import org.apache.http.client.fluent.Request;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.hamcrest.Matcher;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.kitesdk.examples.logging.CreateDataset;
import org.kitesdk.examples.logging.DeleteDataset;
import org.kitesdk.examples.logging.ReadDataset;
import org.kitesdk.minicluster.FlumeService;
import org.kitesdk.minicluster.HdfsService;
import org.kitesdk.minicluster.HiveService;
import org.kitesdk.minicluster.MiniCluster;
import static org.hamcrest.CoreMatchers.any;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.matchers.JUnitMatchers.containsString;

  public static void stopCluster() {
    try {
      cluster.stop();
    } catch (Exception e) {
      // ignore problems during shutdown
    }
  }

  private static void configureLog4j() throws Exception {
    // configuration is done programmatically and not in log4j.properties so that so we
    // can defer initialization to after the Flume Avro RPC source port is running
    Log4jAppender appender = new Log4jAppender();
    appender.setName("flume");
    appender.setHostname("localhost");
    appender.setPort(41415);
    appender.setUnsafeMode(true);
    appender.activateOptions();

    Logger.getLogger("org.kitesdk.examples.logging.webapp.LoggingServlet").addAppender(appender);
    Logger.getLogger("org.kitesdk.examples.logging.webapp.LoggingServlet").setLevel(Level.INFO);
  }

  @Test
  public void test() throws Exception {
    run(new CreateDataset());
    for (int i = 1; i <= 10; i++) {
      get("http://localhost:8080/logging-webapp/send?message=Hello-" + i);
    }
    Thread.sleep(40000); // wait for events to be flushed to HDFS
    run(containsString("{\"id\": 10, \"message\": \"Hello-10\"}"), new ReadDataset());
    run(new DeleteDataset());