[hotfix] [avro] Improve AvroUtils to perform reflection lookups only once. This also fixes minor warnings (unchecked casts) and moves the constants into scopes that avoid bridge methods for access outside of the nested classes.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c438e293 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c438e293 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c438e293 Branch: refs/heads/master Commit: c438e293adc541c56cbaec222aa2942969e49c71 Parents: 6dd5e2b Author: Stephan Ewen <[email protected]> Authored: Thu Nov 2 21:12:08 2017 +0100 Committer: Stephan Ewen <[email protected]> Committed: Fri Nov 3 16:40:35 2017 +0100 ---------------------------------------------------------------------- .../flink/api/java/typeutils/AvroUtils.java | 60 ++++++++++++-------- 1 file changed, 35 insertions(+), 25 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/c438e293/flink-core/src/main/java/org/apache/flink/api/java/typeutils/AvroUtils.java ---------------------------------------------------------------------- diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/AvroUtils.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/AvroUtils.java index e19f5fb..2983be0 100644 --- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/AvroUtils.java +++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/AvroUtils.java @@ -35,13 +35,32 @@ import static org.apache.flink.api.java.typeutils.TypeExtractionUtils.hasSupercl */ public abstract class AvroUtils { - protected static final String AVRO_SPECIFIC_RECORD_BASE = "org.apache.avro.specific.SpecificRecordBase"; + private static final String AVRO_KRYO_UTILS = "org.apache.flink.formats.avro.utils.AvroKryoSerializerUtils"; - protected static final String AVRO_GENERIC_RECORD = "org.apache.avro.generic.GenericData$Record"; + private static final AvroUtils INSTANCE = loadAvroKryoUtils(); - private static final String AVRO_KRYO_UTILS = "org.apache.flink.formats.avro.utils.AvroKryoSerializerUtils"; + private static AvroUtils loadAvroKryoUtils() { + // try and load the special AvroUtils from the flink-avro package + try { + Class<?> clazz = Class.forName(AVRO_KRYO_UTILS, false, AvroUtils.class.getClassLoader()); + return clazz.asSubclass(AvroUtils.class).getConstructor().newInstance(); + } catch (ClassNotFoundException e) { + // cannot find the utils, return the default implementation + return new DefaultAvroUtils(); + } catch (Exception e) { + throw new RuntimeException("Could not instantiate " + AVRO_KRYO_UTILS + ".", e); + } + } - private static final String AVRO_GENERIC_DATA_ARRAY = "org.apache.avro.generic.GenericData$Array"; + /** + * Returns either the default {@link AvroUtils} which throw an exception in cases where Avro + * would be needed or loads the specific utils for Avro from flink-avro. + */ + public static AvroUtils getAvroUtils() { + return INSTANCE; + } + + // ------------------------------------------------------------------------ /** * Loads the utility class from <code>flink-avro</code> and adds Avro-specific serializers. This @@ -66,39 +85,30 @@ public abstract class AvroUtils { */ public abstract <T> TypeInformation<T> createAvroTypeInfo(Class<T> type); + // ------------------------------------------------------------------------ + /** - * Returns either the default {@link AvroUtils} which throw an exception in cases where Avro - * would be needed or loads the specific utils for Avro from flink-avro. + * A default implementation of the AvroUtils used in the absence of Avro. */ - public static AvroUtils getAvroUtils() { + private static class DefaultAvroUtils extends AvroUtils { - // try and load the special AvroUtils from the flink-avro package - Class<?> clazz; - try { - clazz = Class.forName(AVRO_KRYO_UTILS, false, AvroUtils.class.getClassLoader()); - } catch (ClassNotFoundException e) { - // cannot find the utils, return the default implementation - return new DefaultAvroUtils(); - } + private static final String AVRO_SPECIFIC_RECORD_BASE = "org.apache.avro.specific.SpecificRecordBase"; - try { - return (AvroUtils) clazz.getConstructor().newInstance(); - } catch (Exception e) { - throw new RuntimeException("Could not instantiate " + AVRO_KRYO_UTILS + "."); - } - } + private static final String AVRO_GENERIC_RECORD = "org.apache.avro.generic.GenericData$Record"; + + private static final String AVRO_GENERIC_DATA_ARRAY = "org.apache.avro.generic.GenericData$Array"; - private static class DefaultAvroUtils extends AvroUtils { @Override public void addAvroSerializersIfRequired(ExecutionConfig reg, Class<?> type) { - if (hasSuperclass(type, AVRO_SPECIFIC_RECORD_BASE) || hasSuperclass( - type, - AVRO_GENERIC_RECORD)) { + if (hasSuperclass(type, AVRO_SPECIFIC_RECORD_BASE) || + hasSuperclass(type, AVRO_GENERIC_RECORD)) { + throw new RuntimeException("Could not load class '" + AVRO_KRYO_UTILS + "'. " + "You may be missing the 'flink-avro' dependency."); } } + @SuppressWarnings({"rawtypes", "unchecked"}) @Override public void addAvroGenericDataArrayRegistration(LinkedHashMap<String, KryoRegistration> kryoRegistrations) { kryoRegistrations.put(AVRO_GENERIC_DATA_ARRAY,
