// Path: src/main/java/org/apache/hadoop/hive/kafka/AvroSchemaGenerator.java
// public class AvroSchemaGenerator {
//   private TypeInfoToSchema typeInfoToSchema;
// 
//   public AvroSchemaGenerator() {
//     this.typeInfoToSchema = new TypeInfoToSchema();
//   }
// 
//   public Schema getSchema(String columnNamesStr, String columnTypesStr,
//                           String columnCommentsStr, String namespace, String name,
//                           String doc) {
//     List<String> columnNames = Arrays.asList(columnNamesStr.split(","));
//     List<TypeInfo> columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypesStr);
//     List<String> columnComments;
//     if (columnCommentsStr.isEmpty()) {
//       columnComments = new ArrayList<String>();
//     } else {
//       columnComments = Arrays.asList(columnCommentsStr.split(","));
//     }
// 
//     return typeInfoToSchema.convert(columnNames, columnTypes, columnComments, namespace, name, doc);
//   }
// }
// 
// Path: src/main/java/org/apache/hadoop/hive/kafka/KafkaBackedTableProperties.java
// public class KafkaBackedTableProperties {
//   private static final Logger LOG = Logger.getLogger(KafkaBackedTableProperties.class);
//   public static final String KAFKA_URI = "kafka.service.uri";
//   public static final String KAFKA_URL = "kafka.service.url";
//   public static final String KAFKA_PORT = "kafka.service.port";
//   public static final String KAFKA_WHITELIST_TOPICS = "kafka.whitelist.topics";
//   public static final String KAFKA_AVRO_SCHEMA_FILE = "kafka.avro.schema.file";
//   public static final String COLUMN_NAMES = "columns";
//   public static final String COLUMN_TYPES = "columns.types";
//   public static final String COLUMN_COMMENTS = "columns.comments";
// 
//   /*
//    * This method initializes properties of the external table and populates
//    * the jobProperties map with them so that they can be used throughout the job.
//    */
//   public void initialize(Properties tableProperties, Map<String,String> jobProperties,
//                          TableDesc tableDesc) {
// 
//     // Set kafka.whitelist.topics in the jobProperty
//     String kafkaWhitelistTopics = tableProperties.getProperty(KAFKA_WHITELIST_TOPICS);
//     LOG.debug("Kafka whitelist topics : " + kafkaWhitelistTopics);
//     jobProperties.put(KAFKA_WHITELIST_TOPICS, kafkaWhitelistTopics);
// 
//     // Set kafka.avro.schema.file in the jobProperty
//     String kafkaAvroSchemaFile = tableProperties.getProperty(KAFKA_AVRO_SCHEMA_FILE, null);
//     if (kafkaAvroSchemaFile != null) {
//       LOG.debug("Kafka avro schema file : " + kafkaAvroSchemaFile);
//       jobProperties.put(KAFKA_AVRO_SCHEMA_FILE, kafkaAvroSchemaFile);
//     }
// 
//     // Set kafka.url and kafka.port in the jobProperty
//     String kafkaUri = tableProperties.getProperty(KAFKA_URI);
//     LOG.debug("Kafka URI : " + kafkaUri);
//     jobProperties.put(KAFKA_URI, kafkaUri);
//     final String[] uriSplits = kafkaUri.split(":");
//     String kafkaUrl = uriSplits[0];
//     String kafkaPort = uriSplits[1];
//     LOG.debug("Kafka URL : " + kafkaUrl);
//     jobProperties.put(KAFKA_URL, kafkaUrl);
//     LOG.debug("Kafka PORT : " + kafkaPort);
//     jobProperties.put(KAFKA_PORT, kafkaPort);
// 
//     // Set column names in the jobProperty
//     String columnNames = tableProperties.getProperty(COLUMN_NAMES);
//     LOG.debug("Column Names : " + columnNames);
//     jobProperties.put(COLUMN_NAMES, columnNames);
// 
//     // Set column types in the jobProperty
//     String columnTypes = tableProperties.getProperty(COLUMN_TYPES);
//     LOG.debug("Column Types : " + columnTypes);
//     jobProperties.put(COLUMN_TYPES, columnTypes);
// 
//     // Set column types in the jobProperty
//     String columnComments = tableProperties.getProperty(COLUMN_COMMENTS);
//     LOG.debug("Column Comments : " + columnComments);
//     jobProperties.put(COLUMN_COMMENTS, columnComments);
//   }
// }


import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData.Record;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.kafka.AvroSchemaGenerator;
import org.apache.hadoop.hive.kafka.KafkaBackedTableProperties;
import org.apache.hadoop.hive.serde2.avro.AvroGenericRecordWritable;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.util.Properties;
package org.apache.hadoop.hive.kafka.camus;



public class KafkaAvroMessageDecoder extends MessageDecoder<byte[], Record> {
	protected DecoderFactory decoderFactory;
	protected SchemaRegistry<Schema> registry;
	private Schema latestSchema;
  private static Logger log = null;
	
	@Override
	public void init(Properties props, String topicName) {
    super.init(props, topicName);
    if (log == null) {
      log = Logger.getLogger(getClass());
    }

    try {
      SchemaRegistry<Schema> registry = (SchemaRegistry<Schema>) Class
          .forName("org.apache.hadoop.hive.kafka.camus" +
              ".MemorySchemaRegistry").newInstance();

      registry.init(props);

      this.registry = new CachedSchemaRegistry<Schema>(registry);

      Schema.Parser parser = new Schema.Parser();
      Schema schema;

// Path: src/main/java/org/apache/hadoop/hive/kafka/AvroSchemaGenerator.java
// public class AvroSchemaGenerator {
//   private TypeInfoToSchema typeInfoToSchema;
// 
//   public AvroSchemaGenerator() {
//     this.typeInfoToSchema = new TypeInfoToSchema();
//   }
// 
//   public Schema getSchema(String columnNamesStr, String columnTypesStr,
//                           String columnCommentsStr, String namespace, String name,
//                           String doc) {
//     List<String> columnNames = Arrays.asList(columnNamesStr.split(","));
//     List<TypeInfo> columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypesStr);
//     List<String> columnComments;
//     if (columnCommentsStr.isEmpty()) {
//       columnComments = new ArrayList<String>();
//     } else {
//       columnComments = Arrays.asList(columnCommentsStr.split(","));
//     }
// 
//     return typeInfoToSchema.convert(columnNames, columnTypes, columnComments, namespace, name, doc);
//   }
// }
// 
// Path: src/main/java/org/apache/hadoop/hive/kafka/KafkaBackedTableProperties.java
// public class KafkaBackedTableProperties {
//   private static final Logger LOG = Logger.getLogger(KafkaBackedTableProperties.class);
//   public static final String KAFKA_URI = "kafka.service.uri";
//   public static final String KAFKA_URL = "kafka.service.url";
//   public static final String KAFKA_PORT = "kafka.service.port";
//   public static final String KAFKA_WHITELIST_TOPICS = "kafka.whitelist.topics";
//   public static final String KAFKA_AVRO_SCHEMA_FILE = "kafka.avro.schema.file";
//   public static final String COLUMN_NAMES = "columns";
//   public static final String COLUMN_TYPES = "columns.types";
//   public static final String COLUMN_COMMENTS = "columns.comments";
// 
//   /*
//    * This method initializes properties of the external table and populates
//    * the jobProperties map with them so that they can be used throughout the job.
//    */
//   public void initialize(Properties tableProperties, Map<String,String> jobProperties,
//                          TableDesc tableDesc) {
// 
//     // Set kafka.whitelist.topics in the jobProperty
//     String kafkaWhitelistTopics = tableProperties.getProperty(KAFKA_WHITELIST_TOPICS);
//     LOG.debug("Kafka whitelist topics : " + kafkaWhitelistTopics);
//     jobProperties.put(KAFKA_WHITELIST_TOPICS, kafkaWhitelistTopics);
// 
//     // Set kafka.avro.schema.file in the jobProperty
//     String kafkaAvroSchemaFile = tableProperties.getProperty(KAFKA_AVRO_SCHEMA_FILE, null);
//     if (kafkaAvroSchemaFile != null) {
//       LOG.debug("Kafka avro schema file : " + kafkaAvroSchemaFile);
//       jobProperties.put(KAFKA_AVRO_SCHEMA_FILE, kafkaAvroSchemaFile);
//     }
// 
//     // Set kafka.url and kafka.port in the jobProperty
//     String kafkaUri = tableProperties.getProperty(KAFKA_URI);
//     LOG.debug("Kafka URI : " + kafkaUri);
//     jobProperties.put(KAFKA_URI, kafkaUri);
//     final String[] uriSplits = kafkaUri.split(":");
//     String kafkaUrl = uriSplits[0];
//     String kafkaPort = uriSplits[1];
//     LOG.debug("Kafka URL : " + kafkaUrl);
//     jobProperties.put(KAFKA_URL, kafkaUrl);
//     LOG.debug("Kafka PORT : " + kafkaPort);
//     jobProperties.put(KAFKA_PORT, kafkaPort);
// 
//     // Set column names in the jobProperty
//     String columnNames = tableProperties.getProperty(COLUMN_NAMES);
//     LOG.debug("Column Names : " + columnNames);
//     jobProperties.put(COLUMN_NAMES, columnNames);
// 
//     // Set column types in the jobProperty
//     String columnTypes = tableProperties.getProperty(COLUMN_TYPES);
//     LOG.debug("Column Types : " + columnTypes);
//     jobProperties.put(COLUMN_TYPES, columnTypes);
// 
//     // Set column types in the jobProperty
//     String columnComments = tableProperties.getProperty(COLUMN_COMMENTS);
//     LOG.debug("Column Comments : " + columnComments);
//     jobProperties.put(COLUMN_COMMENTS, columnComments);
//   }
// }


// Path: src/main/java/org/apache/hadoop/hive/kafka/camus/KafkaAvroMessageDecoder.java
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData.Record;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.kafka.AvroSchemaGenerator;
import org.apache.hadoop.hive.kafka.KafkaBackedTableProperties;
import org.apache.hadoop.hive.serde2.avro.AvroGenericRecordWritable;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.util.Properties;

package org.apache.hadoop.hive.kafka.camus;



public class KafkaAvroMessageDecoder extends MessageDecoder<byte[], Record> {
	protected DecoderFactory decoderFactory;
	protected SchemaRegistry<Schema> registry;
	private Schema latestSchema;
  private static Logger log = null;
	
	@Override
	public void init(Properties props, String topicName) {
    super.init(props, topicName);
    if (log == null) {
      log = Logger.getLogger(getClass());
    }

    try {
      SchemaRegistry<Schema> registry = (SchemaRegistry<Schema>) Class
          .forName("org.apache.hadoop.hive.kafka.camus" +
              ".MemorySchemaRegistry").newInstance();

      registry.init(props);

      this.registry = new CachedSchemaRegistry<Schema>(registry);

      Schema.Parser parser = new Schema.Parser();
      Schema schema;

      final String schemaFile = props.getProperty(KafkaBackedTableProperties