Я хочу фильтровать свои сообщения по времени их ts_ms. Проблема в том, что я не могу получить ts_ms из сообщений avro. Это мой урезанный avro .avs c файл:
{
"type": "record",
"name": "Envelope",
"namespace": "mysql.company.scores",
"fields": [
{
"name": "before",
"type": [
"null",
{
"type": "record",
"name": "Value",
"fields": [
<Some fields based on scores table>
],
"connect.name": "mysql.company.scores.Value"
}
],
"default": null
},
{
"name": "after",
"type": [
"null",
"Value"
],
"default": null
},
{
"name": "source",
"type": {
"type": "record",
"name": "Source",
"namespace": "io.debezium.connector.mysql",
"fields": [
{
"name": "version",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "connector",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "name",
"type": "string"
},
{
"name": "server_id",
"type": "long"
},
{
"name": "ts_sec",
"type": "long"
},
{
"name": "gtid",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "file",
"type": "string"
},
{
"name": "pos",
"type": "long"
},
{
"name": "row",
"type": "int"
},
{
"name": "snapshot",
"type": [
{
"type": "boolean",
"connect.default": false
},
"null"
],
"default": false
},
{
"name": "thread",
"type": [
"null",
"long"
],
"default": null
},
{
"name": "db",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "table",
"type": [
"null",
"string"
],
"default": null
},
{
"name": "query",
"type": [
"null",
"string"
],
"default": null
}
],
"connect.name": "io.debezium.connector.mysql.Source"
}
},
{
"name": "op",
"type": "string"
},
{
"name": "ts_ms",
"type": [
"null",
"long"
],
"default": null
}
],
"connect.name": "mysql.company.scores.Envelope"
}
Я могу получить доступ до или после, но когда я могу использовать следующий метод с getTs_ms, я получаю символ не могу найти метод:
private boolean isRecordNew(mysql.company.scores.Envelope value){
return value.getTs_ms() > 1580988600000L;
}
Это релевантная часть моего класса serde:
public static Serde<mysql.company.scores.Envelope> getEnvelopeSerde() {
SpecificAvroSerde<mysql.company.scores.Envelope> scoreSerde = new SpecificAvroSerde();
scoreSerde.configure(
Collections.singletonMap(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG,
schemaRegistryUrl), false);
return scoreSerde;
}
Могу ли я получить доступ к полю ts_ms с тем же классом serde, или я должен изменить его, чтобы включить его в значении?