Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit bac1858 (Fri Jul 17 07:20:46 UTC 2020)

Warnings:

• 1 pom.xml files were touched: Check for build and licensing issues.
• No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

• ❓ 1. The [description] looks good.
• ❓ 2. There is [consensus] that the contribution should go into to Flink.
• ❓ 3. Needs [attention] from.
• ❓ 4. The change fits into the overall [architecture].
• ❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

CI report:

• d1e4ba7 Azure: PENDING
• f0da3ce Azure: PENDING
• 9104e12 Azure: PENDING
• 9b557c7 Azure: PENDING
• 71218ee Azure: PENDING
• 52518ee Azure: SUCCESS

+1 for SR and please tag me when you open a documentation PR for this feature

+1 for SR and please tag me when you open a documentation PR for this feature

Sure， thanks for taking care the document.

After a quick glimpse. Could we unify the AvroRowDataDeserializationSchema with ConfluentRegistryAvroRowDataDeserializationSchema and RegistryAvroRowDataSerializationSchema?
I really believe we need just a single AvroRowDeserializationSchema for avro for table API.

I am quite sure sth like this would work:

RowDataDeserializationSchema extends ResultTypeQueryable {
  	private final DeserializationSchema<GenericRecord> nestedSchema;
	private final DeserializationRuntimeConverter runtimeConverter;
	private final TypeInformation<RowData> resultType;

	public AvroRowDataDeserializationSchema2(
			DeserializationSchema<GenericRecord> nestedSchema,
			DeserializationRuntimeConverter runtimeConverter,
			TypeInformation<RowData> resultType) {
		this.nestedSchema = nestedSchema;
		this.runtimeConverter = runtimeConverter;
		this.resultType = resultType;
	}

	@Override
	public void open(InitializationContext context) throws Exception {
		nestedSchema.open(context);
	}

	@Override
	public RowData deserialize(byte[] message) throws IOException {
		try {
			GenericRecord deserialize = nestedSchema.deserialize(message);
			return (RowData) runtimeConverter.convert(deserialize);
		} catch (Exception e) {
			throw new IOException("Failed to deserialize Avro record.", e);
		}
	}

       	@Override
	public boolean isEndOfStream(RowData nextElement) {
		return false;
	}

	@Override
	public TypeInformation<RowData> getProducedType() {
		return resultType;
	}
}

and then you would use it like this:

in AvroFormatFactory:

new RowDataDeserializationSchema(
	AvroDeserializationSchema.forGeneric(AvroSchemaConverter.convertToSchema(rowType)),
        createRowConverter(rowType), // we would need to move this method to some utils class or to a common abstract class for factories
	rowDataTypeInfo
);

in RegistryAvroFormatFactory:

new RowDataDeserializationSchema(
	ConfluentRegistryAvroDeserializationSchema.forGeneric(
		AvroSchemaConverter.convertToSchema(rowType),
		schemaRegistryURL
	),
        createRowConverter(rowType),
	rowDataTypeInfo
);

Thanks for the nice review, i have addressed your comments.

Would flink-avro-confluent be a better module name than flink-avro-confluent-registry? IMO registry has nothing to do with the format itself.

Would flink-avro-confluent be a better module name than flink-avro-confluent-registry? IMO registry has nothing to do with the format itself.

It depends on how we understand it, the confluent avro format is mainly designed for schema registry.

Schema registry is a terminology and people always calls "schema registry url"[1], the same for "schema registry subject".
[1] https://docs.confluent.io/current/schema-registry/index.html#high-availability-for-single-primary-setup

@danny0405 @dawidwys
Any reasons all the fields read and written by this format has prefix 'record_' ? (I'm using flink sql for this client)
I found responsible code probably here but still have problem with this solution:

@danny0405 @dawidwys
Any reasons all the fields read and written by this format has prefix 'record_' ? (I'm using flink sql for this client)
I found responsible code probably here but still have problem with this solution:

flink/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/AvroSchemaConverter.java

Line 365 in de87a2d

String fieldName = rowName + "_" + fieldNames.get(i);

It's because of the current strategy to infer the Avro schema is convert from the CREATE TABLE DDL, and there is no way to get the record name here. So we put a constant record as a prefix. The record write out all have explicit field name, but the type should be compatible.

@danny0405
Problem is this is not compatible. I'm unable to read anything from Kafka using Confluent Registry. Example:
I have data in Kafka with following value schema:

{
  "type": "record",
  "name": "myrecord",
  "fields": [
    {
      "name": "f1",
      "type": "string"
    }
  ]
}

I'm creating table using this avro-confluent format:

create table `test` (
	`f1` STRING
) WITH (
  'connector' = 'kafka', 
  'topic' = 'test', 
  'properties.bootstrap.servers' = 'localhost:9092', 
  'properties.group.id' = 'test1234', 
   'scan.startup.mode' = 'earliest-offset', 
  'format' = 'avro-confluent'
  'avro-confluent.schema-registry.url' = 'http://localhost:8081'
);

When trying to select data I'm getting error:

SELECT * FROM test;
[ERROR] Could not execute SQL statement. Reason:
org.apache.avro.AvroTypeException: Found myrecord, expecting record, missing required field record_f1

Thanks, i think this is a bug, i have logged an issue there. See https://issues.apache.org/jira/browse/FLINK-19779

I have fired a fix https://github.com/apache/flink/pull/13763/files, can you help check if possible @maver1ck :)

I will check... mvn is running

OK. It's working. I'm able to read data.

@danny0405
I think we have one more problem.
When Flink is creating schema in registry nullability is not properly set for logical types.
Examples. Table:

create table `test_logical_null` (
	`string_field` STRING,
	`timestamp_field` TIMESTAMP(3)
) WITH (
  'connector' = 'kafka', 
  'topic' = 'test-logical-null', 
  'properties.bootstrap.servers' = 'localhost:9092', 
  'properties.group.id' = 'test12345', 
   'scan.startup.mode' = 'earliest-offset', 
  'format' = 'avro-confluent', -- Must be set to 'avro-confluent' to configure this format.
  'avro-confluent.schema-registry.url' = 'http://localhost:8081', -- URL to connect to Confluent Schema Registry
  'avro-confluent.schema-registry.subject' = 'test-logical-null' -- Subject name to write to the Schema Registry service; required for sinks
)

Schema:

{
  "type": "record",
  "name": "record",
  "fields": [
    {
      "name": "string_field",
      "type": [
        "string",
        "null"
      ]
    },
    {
      "name": "timestamp_field",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    }
  ]
}

For not null fields:

create table `test_logical_notnull` (
	`string_field` STRING NOT NULL,
	`timestamp_field` TIMESTAMP(3) NOT NULL
) WITH (
  'connector' = 'kafka', 
  'topic' = 'test-logical-notnull', 
  'properties.bootstrap.servers' = 'localhost:9092', 
  'properties.group.id' = 'test12345', 
   'scan.startup.mode' = 'earliest-offset', 
  'format' = 'avro-confluent', -- Must be set to 'avro-confluent' to configure this format.
  'avro-confluent.schema-registry.url' = 'http://localhost:8081', -- URL to connect to Confluent Schema Registry
  'avro-confluent.schema-registry.subject' = 'test-logical-notnull-value' -- Subject name to write to the Schema Registry service; required for sinks
);

Schema

{
  "type": "record",
  "name": "record",
  "fields": [
    {
      "name": "string_field",
      "type": "string"
    },
    {
      "name": "timestamp_field",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    }
  ]
}

As you can see for string_field we have proper union with null (for nullable field). For timestamp_field in both examples union is missing.

EDIT: I added bug report for this
https://issues.apache.org/jira/browse/FLINK-19786

@maver1ck , you are right, we ignore the nullability of TIMESTAMP_WITHOUT_TIME_ZONE, DATE, and TIME_WITHOUT_TIME_ZONE and Decimal, would fix it altogether in this PR.

Thanks @danny0405

I have updated the fix, @maver1ck , please check if you have time.

@danny0405 I see code review is still in progress.
Could you please let me know when this PR would be polished a little ?