Kafka Connect Avro Support Part 1: AvroSchemaManager #687
Conversation
|
|
cdc/sink/schema_registry.go
Outdated
| uri := m.registryURL + "/subjects/" + url.QueryEscape(tableNameToSchemaSubject(tableName)) + "/versions/latest" | ||
| log.Debug("Querying for latest schema", zap.String("uri", uri)) | ||
|
|
||
| req, err := http.NewRequest("GET", uri, nil) |
There was a problem hiding this comment.
Please leave a TODO about TLS support.
cdc/sink/schema_registry.go
Outdated
| zap.Int("status", resp.StatusCode), | ||
| zap.String("uri", uri), | ||
| zap.ByteString("requestBody", payload), | ||
| zap.ByteString("responseBody", body)) |
There was a problem hiding this comment.
Does it contain sensitive data? Eg. password or userdata.
There was a problem hiding this comment.
If the user configures a registry URL that itself contains sensitive data, e.g. http://username:password@example.com, then it is possible. Other data are strictly about the schema, so they should not contain user data.
cdc/sink/schema_registry.go
Outdated
| } | ||
|
|
||
| if jsonResp.ID == 0 { | ||
| return errors.New(fmt.Sprintf("Illegal schema ID returned from Registry %d", jsonResp.ID)) |
There was a problem hiding this comment.
| return errors.New(fmt.Sprintf("Illegal schema ID returned from Registry %d", jsonResp.ID)) | |
| return errors.Errorf("Illegal schema ID returned from Registry %d", jsonResp.ID) |
cdc/sink/schema_registry.go
Outdated
| log.Info("Registered schema successfully", | ||
| zap.Int("id", jsonResp.ID), | ||
| zap.String("uri", uri), | ||
| zap.ByteString("body", body)) |
There was a problem hiding this comment.
ByteString is not readable it produces something like "\u00FF". What about "zap.String"?
There was a problem hiding this comment.
ByteString is readable when it contains only printable characters, and per my understanding, a JSON body should not contain any unprintable characters. If it does, then something suspicious is definitely happening, so some log output is necessary.
cdc/sink/schema_registry_test.go
Outdated
| }, | ||
| { | ||
| "type": [ | ||
| "null", |
There was a problem hiding this comment.
| "null", | |
| "null", |
Please avoid mixing tab and space.
cdc/sink/schema_registry.go
Outdated
| var regexRemoveSpaces = regexp.MustCompile(`\s`) | ||
|
|
||
| // Register the latest schema for a table to the Registry, by passing in a Codec | ||
| func (m *AvroSchemaManager) Register(tableName model.TableName, codec *goavro.Codec) error { |
There was a problem hiding this comment.
Is there any go package handles registering schema?
|
Please fix data race in the unit test |
|
/run-integration-tests |
Codecov Report
@@ Coverage Diff @@
## master #687 +/- ##
===========================================
Coverage 32.3914% 32.3914%
===========================================
Files 91 91
Lines 9555 9555
===========================================
Hits 3095 3095
Misses 6196 6196
Partials 264 264 |
What problem does this PR solve?
What is changed and how it works?
Check List
Tests
Code changes
Release note
This is the first PR that will lead to support for outputting Avro-serialized data to Kafka. See more in #660. After research and deliberation, we gave up on syncing the Registry Schema ID through Etcd.