import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public T deserialize(String topic, byte[] data) {
  LOGGER.debug("data to deserialize='{}'", DatatypeConverter.printHexBinary(data));
  try {
    // get the schema
    Schema schema = targetType.newInstance().getSchema();

    Injection<GenericRecord, byte[]> genericRecordInjection = GenericAvroCodecs.toBinary(schema);
    GenericRecord genericRecord = genericRecordInjection.invert((byte[]) data).get();
    T result = (T) SpecificData.get().deepCopy(schema, genericRecord);

    LOGGER.debug("data='{}'", result);
    return result;
  } catch (Exception e) {
    throw new SerializationException(
        "Can't deserialize data [" + Arrays.toString(data) + "] from topic [" + topic + "]", e);
  }
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 *
 */
@Override
public void run() {
  Properties props = HopsUtil.getKafkaProperties().getConsumerConfig();
  consumer = new KafkaConsumer<>(props);
  //Subscribe to the Kafka topic
  consumer.subscribe(Collections.singletonList(topic));
  while (consume) {
    ConsumerRecords<Integer, String> records = consumer.poll(1000);
    //Get the records
    for (ConsumerRecord<Integer, String> record : records) {
      //Convert the record using the schema
      Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);
      GenericRecord genericRecord = recordInjection.invert(record.value().getBytes()).get();
      LOGGER.log(Level.INFO, "Consumer received message:{0}", genericRecord);
    }
  }
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 * 
 * @param t
 * @return 
 */
@Override
public byte[] serialize(Tuple4<String, String, String, String> t) {

  if (!initialized) {
    parser = new Schema.Parser();
    schema = parser.parse(schemaJson);
    recordInjection = GenericAvroCodecs.toBinary(schema);
    initialized = true;
  }
  GenericData.Record avroRecord = new GenericData.Record(schema);
  for (int i = 0; i < t.getArity() - 1; i += 2) {
    avroRecord.put(t.getField(i).toString(), t.getField(i + 1).toString());
  }

  byte[] bytes = recordInjection.apply(avroRecord);
  return bytes;
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
public void run() {
	try {
		ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
		Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(User.getClassSchema());

		while (it.hasNext()) {
			MessageAndMetadata<byte[], byte[]> messageAndMetadata = it.next();
			String key = new String(messageAndMetadata.key());
			User user = genericRecordToUser(recordInjection.invert(messageAndMetadata.message()).get());
			// User user = (User)
			// recordInjection.invert(messageAndMetadata.message()).get();
			String summary = "Thread " + m_threadNumber + ", topic=" + messageAndMetadata.topic() + ", partition="
					+ messageAndMetadata.partition() + ", key=" + key + ", user=" + user.toString() + ", offset="
					+ messageAndMetadata.offset() + ", timestamp=" + messageAndMetadata.timestamp()
					+ ", timestampType=" + messageAndMetadata.timestampType();
			System.out.println(summary);
		}
		System.out.println("Shutting down Thread: " + m_threadNumber);
	} catch (Exception e) {
		System.out.println("Exception in thread "+m_threadNumber);
		System.out.println(e);
		e.printStackTrace();
	}
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 *
 * @param args
 * @throws InterruptedException
 */
public static void main(String[] args) throws InterruptedException {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

    Schema.Parser parser = new Schema.Parser();
    Schema schema = parser.parse(USER_SCHEMA);
    Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);

    KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);
    SplittableRandom random = new SplittableRandom();

    while (true) {
        GenericData.Record avroRecord = new GenericData.Record(schema);
        avroRecord.put("str1", "Str 1-" + random.nextInt(10));
        avroRecord.put("str2", "Str 2-" + random.nextInt(1000));
        avroRecord.put("int1", random.nextInt(10000));

        byte[] bytes = recordInjection.apply(avroRecord);

        ProducerRecord<String, byte[]> record = new ProducerRecord<>("mytopic", bytes);
        producer.send(record);
        Thread.sleep(100);
    }

}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 *
 * @param topic
 * @param schema
 * @param lingerDelay
 */
public DelaProducer(String topic, Schema schema, long lingerDelay) {
  super(HopsProcessType.PRODUCER, topic, schema);
  Properties props = HopsUtil.getKafkaProperties().defaultProps();
  props.put(ProducerConfig.CLIENT_ID_CONFIG, "DelaProducer");
  props.put(ProducerConfig.LINGER_MS_CONFIG, lingerDelay);
  producer = new KafkaProducer<>(props);
  recordInjection = GenericAvroCodecs.toBinary(schema);
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 * Create a Producer to stream messages to Kafka.
 *
 * @param topic
 * @param userProps
 * @throws SchemaNotFoundException
 * @throws io.hops.util.CredentialsNotFoundException
 */
public HopsProducer(String topic, Properties userProps) throws SchemaNotFoundException, CredentialsNotFoundException {
  super(HopsProcessType.PRODUCER, topic);
  Properties props = HopsUtil.getKafkaProperties().defaultProps();
  props.put(ProducerConfig.CLIENT_ID_CONFIG, "HopsProducer");
  if (userProps != null) {
    props.putAll(userProps);
  }
  producer = new KafkaProducer<>(props);
  recordInjection = GenericAvroCodecs.toBinary(schema);
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 * 
 * @param bytes
 * @return
 * @throws IOException 
 */
@Override
public String deserialize(byte[] bytes) throws IOException {
  if (!initialized) {
    parser = new Schema.Parser();
    schema = parser.parse(schemaJson);
    recordInjection = GenericAvroCodecs.toBinary(schema);
    initialized = true;
  }
  GenericRecord genericRecord = recordInjection.invert(bytes).get();
  return genericRecord.toString().replaceAll("\\\\u001A", "");
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 * Get record Injections for Avro messages, using topics extracted from Spark
 * HopsWorks Jobs UI.
 *
 * @return
 */
public static Map<String, Injection<GenericRecord, byte[]>> getRecordInjections() {
  Map<String, Schema> schemas = HopsUtil.getSchemas();
  if (schemas.isEmpty()) {
    return null;
  }
  Map<String, Injection<GenericRecord, byte[]>> recordInjections
      = new HashMap<>();
  for (String topic : HopsUtil.getTopics()) {
    recordInjections.
        put(topic, GenericAvroCodecs.toBinary(schemas.get(topic)));

  }
  return recordInjections;
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
public static void main(String[] args) throws InterruptedException {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

    Schema.Parser parser = new Schema.Parser();
    Schema schema = parser.parse(USER_SCHEMA);
    Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);

    KafkaProducer<String, byte[]> producer = new KafkaProducer<>(props);
    for (int i = 0; i < 1000; i++) {
        GenericData.Record avroRecord = new GenericData.Record(schema);
        avroRecord.put("str1", "Str 1-" + i);
        avroRecord.put("str2", "Str 2-" + i);
        avroRecord.put("int1", i);

        byte[] bytes = recordInjection.apply(avroRecord);

        ProducerRecord<String, byte[]> record = new ProducerRecord<>("mytopic", bytes);
        producer.send(record);

        Thread.sleep(250);
    }

    producer.close();
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 * The main method.
 *
 * @param args the arguments
 */
public static void main(String[] args) {
	LogTool.initConsole();
    long events = Long.parseLong(args[0]);
    Random rnd = new Random();
 
    Properties props = new Properties();
    //props.put("metadata.broker.list", "hp-lt-ubuntu-1:9092");
    props.put("bootstrap.servers", "hp-lt-ubuntu-1:9092");
    props.put("producer.type", "sync");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
    //props.put("partition.assignment.strategy", "range");
    //props.put("partitioner.class", "com.pzybrick.kafka1.training.KafkaPartitionerTest");
    props.put("request.required.acks", "1");
    props.put("group.id", "group1");
 
    //Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(User.getClassSchema());
    Injection<GenericRecord, byte[]> recordInjectionUser = GenericAvroCodecs.toBinary(User.getClassSchema());
    
    KafkaProducer<String, byte[]> producer = new KafkaProducer<String, byte[]>(props);
    long keyNum = System.currentTimeMillis();
 
    for (long nEvents = 0; nEvents < events; nEvents++) { 
    	   System.out.println("creating event "+nEvents);
           String key = String.valueOf(keyNum);
           
           User user = User.newBuilder().setName("name"+keyNum).setFavoriteColor("color"+keyNum).setFavoriteNumber((int)keyNum).build();
           byte[] bytes = recordInjectionUser.apply(user);
           ProducerRecord<String, byte[]> data = new ProducerRecord<String, byte[]>("pzmultiptn1", key, bytes);
           producer.send(data);
           keyNum++;
    }
    producer.close();
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
@Override
public byte[] serialize(String topic, T data) {
  LOGGER.debug("data to serialize='{}'", data);

  Injection<GenericRecord, byte[]> genericRecordInjection =
      GenericAvroCodecs.toBinary(data.getSchema());
  byte[] result = genericRecordInjection.apply(data);

  LOGGER.debug("serialized data='{}'", DatatypeConverter.printHexBinary(result));
  return result;
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
/**
 * Instantiates a new serialized persisted data.
 *
 * @param uuid the uuid
 * @param datatype
 * @param avroSchema the avro schema
 * @param serializedData the serialized data
 * @param timestamp the timestamp
 */
public SerializedPersistedData(UUID uuid, String datatype, String avroSchema, byte[] serializedData, Date timestamp, UUID freedomoticInstanceId) {
    super();
    this.uuid = uuid;
    this.datatype = datatype;
    this.avroSchema = new Schema.Parser().parse(avroSchema);
    this.serializedData = serializedData;
    this.persistenceTimestamp = timestamp;
    this.recordInjection = GenericAvroCodecs.toBinary(this.avroSchema);
    this.freedomoticInstanceId = freedomoticInstanceId;
}
 

import com.twitter.bijection.avro.GenericAvroCodecs; //导入依赖的package包/类
public static void main(String[] args) {


        Schema schema = User.SCHEMA$;

        Injection<GenericRecord, byte[]> recordInjection = GenericAvroCodecs.toBinary(schema);
        User  user=    new User("1",1,"");

        byte[] bytes = recordInjection.apply(user);
        GenericRecord record = recordInjection.invert(bytes).get();

        Object name = record.get("name");

        System.out.println(record);

        }