You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2017/11/03 17:11:41 UTC
[18/21] flink git commit: [hotfix] [avro] Improve AvroUtils to
perform reflection lookups only once.
[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 <se...@apache.org>
Authored: Thu Nov 2 21:12:08 2017 +0100
Committer: Stephan Ewen <se...@apache.org>
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,