[FLINK-9384] [table] Fix KafkaAvroTableSource type mismatch

This closes apache#6026.
fhueske · May 25, 2018 · bc8d1b1 · bc8d1b1
1 parent cb48019
commit bc8d1b1
Show file tree

Hide file tree

Showing 4 changed files with 37 additions and 12 deletions.
diff --git a/...c/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java b/...c/test/java/org/apache/flink/streaming/connectors/kafka/KafkaAvroTableSourceTestBase.java
@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.formats.avro.utils.AvroTestUtils;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.Types;
 
 import org.apache.avro.Schema;
@@ -72,6 +73,10 @@ public void testSameFieldsAvroClass() {
 
  // check field mapping
  assertNull(source.getFieldMapping());
+
+ // check if DataStream type matches with TableSource.getReturnType()
+ assertEquals(source.getReturnType(),
+ source.getDataStream(StreamExecutionEnvironment.getExecutionEnvironment()).getType());
  }
 
  @Test
@@ -117,6 +122,10 @@ public void testDifferentFieldsAvroClass() {
  assertEquals("otherField1", fieldMapping.get("field1"));
  assertEquals("otherField2", fieldMapping.get("field2"));
  assertEquals("otherField3", fieldMapping.get("field3"));
+
+ // check if DataStream type matches with TableSource.getReturnType()
+ assertEquals(source.getReturnType(),
+ source.getDataStream(StreamExecutionEnvironment.getExecutionEnvironment()).getType());
  }
 
  /**

diff --git a/.../flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java b/.../flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
@@ -18,6 +18,8 @@
 package org.apache.flink.formats.avro;
 
 import org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.formats.avro.typeutils.AvroRecordClassConverter;
 import org.apache.flink.formats.avro.utils.MutableByteArrayInputStream;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
@@ -30,6 +32,7 @@
 import org.apache.avro.specific.SpecificData;
 import org.apache.avro.specific.SpecificDatumReader;
 import org.apache.avro.specific.SpecificRecord;
+import org.apache.avro.specific.SpecificRecordBase;
 import org.apache.avro.util.Utf8;
 
 import java.io.IOException;
@@ -76,19 +79,25 @@ public class AvroRowDeserializationSchema extends AbstractDeserializationSchema<
  */
  private SpecificRecord record;
 
+ /**
+ * Type information describing the result type.
+ */
+ private transient TypeInformation<Row> typeInfo;
+
  /**
  * Creates a Avro deserialization schema for the given record.
  *
  * @param recordClazz Avro record class used to deserialize Avro's record to Flink's row
  */
- public AvroRowDeserializationSchema(Class<? extends SpecificRecord> recordClazz) {
+ public AvroRowDeserializationSchema(Class<? extends SpecificRecordBase> recordClazz) {
  Preconditions.checkNotNull(recordClazz, "Avro record class must not be null.");
  this.recordClazz = recordClazz;
  this.schema = SpecificData.get().getSchema(recordClazz);
  this.datumReader = new SpecificDatumReader<>(schema);
  this.record = (SpecificRecord) SpecificData.newInstance(recordClazz, schema);
  this.inputStream = new MutableByteArrayInputStream();
  this.decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
+ this.typeInfo = AvroRecordClassConverter.convert(recordClazz);
  }
 
  @Override
@@ -120,6 +129,11 @@ private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IO
  this.decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
  }
 
+ @Override
+ public TypeInformation<Row> getProducedType() {
+ return typeInfo;
+ }
+
  /**
  * Converts a (nested) Avro {@link SpecificRecord} into Flink's Row type.
  * Avro's {@link Utf8} fields are converted into regular Java strings.

diff --git a/...nk-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDeSerializationSchemaTest.java b/...nk-avro/src/test/java/org/apache/flink/formats/avro/AvroRowDeSerializationSchemaTest.java
@@ -24,6 +24,7 @@
 import org.apache.flink.util.InstantiationUtil;
 
 import org.apache.avro.specific.SpecificRecord;
+import org.apache.avro.specific.SpecificRecordBase;
 import org.junit.Test;
 
 import java.io.IOException;
@@ -37,7 +38,7 @@ public class AvroRowDeSerializationSchemaTest {
 
  @Test
  public void testSerializeDeserializeSimpleRow() throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData();
 
  final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0);
  final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0);
@@ -50,7 +51,7 @@ public void testSerializeDeserializeSimpleRow() throws IOException {
 
  @Test
  public void testSerializeSimpleRowSeveralTimes() throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData();
 
  final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0);
  final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0);
@@ -65,7 +66,7 @@ public void testSerializeSimpleRowSeveralTimes() throws IOException {
 
  @Test
  public void testDeserializeRowSeveralTimes() throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> testData = AvroTestUtils.getSimpleTestData();
 
  final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0);
  final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0);
@@ -80,7 +81,7 @@ public void testDeserializeRowSeveralTimes() throws IOException {
 
  @Test
  public void testSerializeDeserializeComplexRow() throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
 
  final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0);
  final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0);
@@ -93,7 +94,7 @@ public void testSerializeDeserializeComplexRow() throws IOException {
 
  @Test
  public void testSerializeComplexRowSeveralTimes() throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
 
  final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0);
  final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0);
@@ -108,7 +109,7 @@ public void testSerializeComplexRowSeveralTimes() throws IOException {
 
  @Test
  public void testDeserializeComplexRowSeveralTimes() throws IOException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
 
  final AvroRowSerializationSchema serializationSchema = new AvroRowSerializationSchema(testData.f0);
  final AvroRowDeserializationSchema deserializationSchema = new AvroRowDeserializationSchema(testData.f0);
@@ -123,7 +124,7 @@ public void testDeserializeComplexRowSeveralTimes() throws IOException {
 
  @Test
  public void testSerializability() throws IOException, ClassNotFoundException {
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> testData = AvroTestUtils.getComplexTestData();
 
  final AvroRowSerializationSchema serOrig = new AvroRowSerializationSchema(testData.f0);
  final AvroRowDeserializationSchema deserOrig = new AvroRowDeserializationSchema(testData.f0);

diff --git a/...k-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/utils/AvroTestUtils.java b/...k-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/utils/AvroTestUtils.java
@@ -33,6 +33,7 @@
 import org.apache.avro.io.EncoderFactory;
 import org.apache.avro.reflect.ReflectData;
 import org.apache.avro.specific.SpecificRecord;
+import org.apache.avro.specific.SpecificRecordBase;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
@@ -70,7 +71,7 @@ public static Schema createFlatAvroSchema(String[] fieldNames, TypeInformation[]
  /**
  * Tests a simple Avro data types without nesting.
  */
- public static Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> getSimpleTestData() {
+ public static Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> getSimpleTestData() {
  final Address addr = Address.newBuilder()
  .setNum(42)
  .setStreet("Main Street 42")
@@ -86,7 +87,7 @@ public static Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> getSi
  rowAddr.setField(3, "Test State");
  rowAddr.setField(4, "12345");
 
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> t = new Tuple3<>();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> t = new Tuple3<>();
  t.f0 = Address.class;
  t.f1 = addr;
  t.f2 = rowAddr;
@@ -97,7 +98,7 @@ public static Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> getSi
  /**
  * Tests all Avro data types as well as nested types.
  */
- public static Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> getComplexTestData() {
+ public static Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> getComplexTestData() {
  final Address addr = Address.newBuilder()
  .setNum(42)
  .setStreet("Main Street 42")
@@ -148,7 +149,7 @@ public static Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> getCo
  rowUser.setField(13, null);
  rowUser.setField(14, rowAddr);
 
- final Tuple3<Class<? extends SpecificRecord>, SpecificRecord, Row> t = new Tuple3<>();
+ final Tuple3<Class<? extends SpecificRecordBase>, SpecificRecord, Row> t = new Tuple3<>();
  t.f0 = User.class;
  t.f1 = user;
  t.f2 = rowUser;