import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private static Class<? extends SpecificRecordBuilderBase> getAvroBuilderClass(
    Class<? extends SpecificRecordBase> avroClass) {
  for (Class<?> clazz : avroClass.getClasses()) {
    if (SpecificRecordBuilderBase.class.isAssignableFrom(clazz)) {
      try {
        if (clazz.getMethod("build").getReturnType() != avroClass) {
          continue;
        }
      } catch (NoSuchMethodException expected) {
      }
      return (Class<? extends SpecificRecordBuilderBase>) clazz;
    }
  }
  return null;
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/** This is the place you can access map-reduce workflow node parameters */
@SuppressWarnings("unchecked")
@Override
protected void setup(Context context) throws IOException, InterruptedException {
    String moduleClassName = context.getConfiguration().get(EXPORT_ACTION_BUILDER_FACTORY_CLASSNAME);
    if (StringUtils.isNotBlank(moduleClassName)) {
        try {
            Class<?> clazz = Class.forName(moduleClassName);
            Constructor<?> constructor = clazz.getConstructor();
            ActionBuilderFactory<SpecificRecordBase> actionBuilderFactory = (ActionBuilderFactory<SpecificRecordBase>) constructor.newInstance();
            actionBuilder = actionBuilderFactory.instantiate(context.getConfiguration(), StaticConfigurationProvider.AGENT_DEFAULT,
                    getActionSetId(actionBuilderFactory.getAlgorithName(), context.getConfiguration()));
        } catch (Exception e) {
            throw new RuntimeException(
                    "unexpected exception ocurred when instantiating " + "builder module: " + moduleClassName, e);
        }
    } else {
        throw new InvalidParameterException("unknown action builder module instance, " + "no "
                + EXPORT_ACTION_BUILDER_FACTORY_CLASSNAME + " parameter provided!");
    }
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Build and return properties by specific properties class.
 *
 * @param propertiesClass the class of properties
 * @param <S>             the class of properties
 * @return property instance
 */
public <S extends SpecificRecordBase> S getSpecificProperties(Class<S> propertiesClass) {
  Properties entity = findOrCreateByClass(propertiesClass);
  S specificProperties = null;
  if (entity != null) {
    AvroByteArrayConverter<S> converter =
        new AvroByteArrayConverter<>(propertiesClass);
    try {
      specificProperties = converter.fromByteArray(entity.getRawConfiguration());
    } catch (IOException ex) {
      LOG.error(
          "Unable to parse raw data for specific record " + propertiesClass.getSimpleName(), ex);
    }
  }
  if (specificProperties == null) {
    specificProperties = buildDefaultProperties(propertiesClass);
  }
  return specificProperties;
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
private <S extends SpecificRecordBase> S buildDefaultProperties(Class<S> propertiesClass) {
  S result = null;
  try {
    Schema schema = (Schema) propertiesClass.getField(SCHEMA).get(null);
    RawSchema rawSchema = new RawSchema(schema.toString());
    DefaultRecordGenerationAlgorithm<RawData> algotithm =
        new DefaultRecordGenerationAlgorithmImpl<>(rawSchema, new RawDataFactory());
    RawData rawData = algotithm.getRootData();
    AvroJsonConverter<S> converter = new AvroJsonConverter<>(schema, propertiesClass);
    result = converter.decodeJson(rawData.getRawData());
  } catch (Exception ex) {
    LOG.error(
        "Unable to build default specific properties for class "
            + propertiesClass.getSimpleName(), ex);
  }
  return result;
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
private <S extends SpecificRecordBase> Properties createDefault(Class<S> propertiesClass) {
  Properties properties = new Properties();

  S specificProperties = buildDefaultProperties(propertiesClass);
  AvroByteArrayConverter<S> converter =
      new AvroByteArrayConverter<>(propertiesClass);
  try {
    properties.setRawConfiguration(converter.toByteArray(specificProperties));
  } catch (IOException ex) {
    LOG.error("Unable to serialize configuration for properties", ex);
  }
  properties.setFqn(propertiesClass.getName());
  Long id = save(properties);
  properties.setId(id);
  return properties;
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Finds all of the messages in the specified packaged and calls register.
 * @param messagePackage A string which contains the full name of the
 *                       package containing the protocol messages.
 */
@Inject
private ProtocolSerializer(
    @Parameter(ProtocolSerializerNamespace.class) final String messagePackage) {

  // Build a list of the message reflection classes.
  final ScanResult scanResult = new FastClasspathScanner(messagePackage).scan();
  final List<String> scanNames = scanResult.getNamesOfSubclassesOf(SpecificRecordBase.class);
  final List<Class<?>> messageClasses = scanResult.classNamesToClassRefs(scanNames);

  // Add the header message from the org.apache.reef.wake.avro.message package.
  messageClasses.add(Header.class);

  // Register all of the messages in the specified package.
  for (final Class<?> cls : messageClasses) {
    this.register(cls);
  }
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
@Override
public void setupModule(SetupContext context) {
  // Serialization is easy - this mixin disables serialization of the Schema field,
  // and everything else Just Works.
  context.setMixInAnnotations(SpecificRecordBase.class, AvroMappingMixin.class);
  // Deserialization is harder. Registering a custom Deserializers instance allows us to manually
  // construct a JsonDeserializer each time the ObjectMapper encounters a new type, so we can
  // detect SpecificRecordBase subclasses and handle them specially.
  context.addDeserializers(new AvroJacksonDeserializers());
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Creates a Kafka 0.9 Avro {@link StreamTableSource} using a given {@link SpecificRecord}.
 *
 * @param topic      Kafka topic to consume.
 * @param properties Properties for the Kafka consumer.
 * @param schema	 Schema of the produced table.
 * @param record     Avro specific record.
 */
public Kafka09AvroTableSource(
	String topic,
	Properties properties,
	TableSchema schema,
	Class<? extends SpecificRecordBase> record) {

	super(
		topic,
		properties,
		schema,
		record);
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Creates a Kafka 0.8 Avro {@link StreamTableSource} using a given {@link SpecificRecord}.
 *
 * @param topic      Kafka topic to consume.
 * @param properties Properties for the Kafka consumer.
 * @param schema     Schema of the produced table.
 * @param record     Avro specific record.
 */
public Kafka08AvroTableSource(
	String topic,
	Properties properties,
	TableSchema schema,
	Class<? extends SpecificRecordBase> record) {

	super(
		topic,
		properties,
		schema,
		record);
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Creates a Kafka 0.11 Avro {@link StreamTableSource} using a given {@link SpecificRecord}.
 *
 * @param topic      Kafka topic to consume.
 * @param properties Properties for the Kafka consumer.
 * @param schema     Schema of the produced table.
 * @param record     Avro specific record.
 */
public Kafka011AvroTableSource(
	String topic,
	Properties properties,
	TableSchema schema,
	Class<? extends SpecificRecordBase> record) {

	super(
		topic,
		properties,
		schema,
		record);
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Creates a Kafka 0.10 Avro {@link StreamTableSource} using a given {@link SpecificRecord}.
 *
 * @param topic      Kafka topic to consume.
 * @param properties Properties for the Kafka consumer.
 * @param schema     Schema of the produced table.
 * @param record     Avro specific record.
 */
public Kafka010AvroTableSource(
	String topic,
	Properties properties,
	TableSchema schema,
	Class<? extends SpecificRecordBase> record) {

	super(
		topic,
		properties,
		schema,
		record);
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Creates a generic Kafka Avro {@link StreamTableSource} using a given {@link SpecificRecord}.
 *
 * @param topic            Kafka topic to consume.
 * @param properties       Properties for the Kafka consumer.
 * @param schema           Schema of the produced table.
 * @param avroRecordClass  Class of the Avro record that is read from the Kafka topic.
 */
protected KafkaAvroTableSource(
	String topic,
	Properties properties,
	TableSchema schema,
	Class<? extends SpecificRecordBase> avroRecordClass) {

	super(
		topic,
		properties,
		schema,
		convertToRowTypeInformation(avroRecordClass));

	this.avroRecordClass = avroRecordClass;
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Converts the extracted AvroTypeInfo into a RowTypeInfo nested structure with deterministic field order.
 * Replaces generic Utf8 with basic String type information.
 */
@SuppressWarnings("unchecked")
private static <T extends SpecificRecordBase> TypeInformation<Row> convertToRowTypeInformation(Class<T> avroClass) {
	final AvroTypeInfo<T> avroTypeInfo = new AvroTypeInfo<>(avroClass);
	// determine schema to retrieve deterministic field order
	final Schema schema = SpecificData.get().getSchema(avroClass);
	return (TypeInformation<Row>) convertToTypeInformation(avroTypeInfo, schema);
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
@Override
public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot) {
	if (configSnapshot instanceof AvroSchemaSerializerConfigSnapshot ||
			configSnapshot instanceof AvroSerializerConfigSnapshot) {

		// avro serializer, nice :-)
		checkState(serializer instanceof AvroSerializer,
				"Serializer was changed backwards to PojoSerializer and now encounters AvroSerializer snapshot.");

		return serializer.ensureCompatibility(configSnapshot);
	}
	else if (configSnapshot instanceof PojoSerializerConfigSnapshot) {
		// common previous case
		checkState(SpecificRecordBase.class.isAssignableFrom(type),
				"BackwardsCompatibleAvroSerializer resuming a state serialized " +
						"via a PojoSerializer, but not for an Avro Specific Record");

		final AvroTypeInfo<? extends SpecificRecordBase> typeInfo =
				new AvroTypeInfo<>(type.asSubclass(SpecificRecordBase.class), true);

		@SuppressWarnings("unchecked")
		final TypeSerializer<T> pojoSerializer =
				(TypeSerializer<T>) typeInfo.createPojoSerializer(new ExecutionConfig());
		this.serializer = pojoSerializer;
		return serializer.ensureCompatibility(configSnapshot);
	}
	else if (configSnapshot instanceof KryoRegistrationSerializerConfigSnapshot) {
		// force-kryo old case common previous case
		// we create a new Kryo Serializer with a blank execution config.
		// registrations are anyways picked up from the snapshot.
		serializer = new KryoSerializer<>(type, new ExecutionConfig());
		return serializer.ensureCompatibility(configSnapshot);
	}
	else {
		// completely incompatible type, needs migration
		return CompatibilityResult.requiresMigration();
	}
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
private static <T extends SpecificRecordBase> List<PojoField> generateFieldsFromAvroSchema(Class<T> typeClass) {
	PojoTypeExtractor pte = new PojoTypeExtractor();
	ArrayList<Type> typeHierarchy = new ArrayList<>();
	typeHierarchy.add(typeClass);
	TypeInformation ti = pte.analyzePojo(typeClass, typeHierarchy, null, null, null);

	if(!(ti instanceof PojoTypeInfo)) {
		throw new IllegalStateException("Expecting type to be a PojoTypeInfo");
	}
	PojoTypeInfo pti =  (PojoTypeInfo) ti;
	List<PojoField> newFields = new ArrayList<>(pti.getTotalFields());

	for(int i = 0; i < pti.getArity(); i++) {
		PojoField f = pti.getPojoFieldAt(i);
		TypeInformation newType = f.getTypeInformation();
		// check if type is a CharSequence
		if(newType instanceof GenericTypeInfo) {
			if((newType).getTypeClass().equals(CharSequence.class)) {
				// replace the type by a org.apache.avro.util.Utf8
				newType = new GenericTypeInfo(org.apache.avro.util.Utf8.class);
			}
		}
		PojoField newField = new PojoField(f.getField(), newType);
		newFields.add(newField);
	}
	return newFields;
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 *
 * @param recordClass		Avro record type class
 * @return					the Avro schema for the supplied class
 * @throws ToolException	if an exception is thrown extracting the schema from the record class
 */
public static Schema getSchema(Class<? extends SpecificRecordBase> recordClass) throws ToolException {
	try {
		// Will have a static member SCHEMA$ from which we can parse the schema:
		Field field = recordClass.getDeclaredField("SCHEMA$");
		return (Schema)field.get(null);
	}
	catch (NoSuchFieldException | SecurityException
			| IllegalArgumentException | IllegalAccessException e) {
		throw new ToolException(e);
	}
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
@Override
protected void map(AvroKey<? extends SpecificRecordBase> key, NullWritable ignore, Context context)
        throws IOException, InterruptedException {
    List<AtomicAction> actions = createActions(key.datum());
    if (actions != null) {
        for (AtomicAction action : actions) {
            Text keyOut = new Text();
            Text valueOut = new Text();
            keyOut.set(action.getRowKey());
            valueOut.set(action.toString());
            context.write(keyOut, valueOut);
        }
    }
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Creates list of actions for given avro object.
 * 
 * @param datum source avro object
 */
private List<AtomicAction> createActions(SpecificRecordBase datum) {
    try {
        return actionBuilder.build(datum);
    } catch (TrustLevelThresholdExceededException e) {
        return Collections.emptyList();
    }
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
public static <T extends SpecificRecordBase> byte[] serialize(T record) throws IOException {
    try {
        DatumWriter<T> writer = new SpecificDatumWriter<T>(record.getSchema());
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
        writer.write(record, encoder);
        encoder.flush();
        IOUtils.closeQuietly(out);
        return out.toByteArray();
    } catch (IOException e) {
        e.printStackTrace();
        return null;
    }
}
 

import org.apache.avro.specific.SpecificRecordBase; //导入依赖的package包/类
/**
 * Found property from database or create new if not found, update it and save to database.
 *
 * @param propertiesDto   the property to update
 * @param propertiesClass the property class
 * @param <S>             the property class
 * @return property dto
 */
public <S extends SpecificRecordBase> PropertiesDto editPropertiesDto(
    PropertiesDto propertiesDto,Class<S> propertiesClass) throws Exception {
  Properties entity = findOrCreateByClass(propertiesClass);
  GenericRecord record = FormAvroConverter.createGenericRecordFromRecordField(
      propertiesDto.getConfiguration());
  GenericAvroConverter<GenericRecord> converter = new GenericAvroConverter<>(record.getSchema());
  byte[] rawConfiguration = converter.encode(record);
  entity.setRawConfiguration(rawConfiguration);
  save(entity);
  return toDto(entity, propertiesClass);
}