はじめに
Confluent Platformを使ってみた時のメモ書きです。
今回はSchema Registryを使用したメッセージ送受信を試してみます。
関連記事
Confluent Platform メモ - (1)環境構築
Confluent Platform メモ - (2)メッセージ送受信簡易テスト
Confluent Platform メモ - (3)Schema Registry簡易テスト
参考情報
Schema Registry Overview
Avro Schema Serializer and Deserializer
Apache Avro™ 1.10.2 Specification
Avro,SchemaRegistryことはじめ
Schema Registryについて
なんだかんだでSchema Registryの動作イメージについては公式ドキュメントの以下の図が分かりやすいと思います。
Schema Registry Overviewからの抜粋
こんな感じの流れになると思います。
メッセージ送信時:
メッセージフォーマットを示すSchema(Avro, JSONなど)を準備します。
ProducerはSchemaに応じてデータをシリアライズします。
使用したSchemaをSchema Registryに登録すると、使用したSchemaを識別するidが返されるので、idとデータをセットでKafkaのTopicに送信します。
つまり、"このSchemaが使われていますよ"という情報(id)と、シリアライズされたデータがメッセージとして送信されることになります。
メッセージ受信時:
Schema idとシリアライズされたデータがセットで受信されるので、Consumer側ではidに紐づくSchemaをSchem Registryから取得してデータをデシリアライズします。
これで、Producer/Consumer両者共に意図したSchema(メッセージフォーマット)でメッセージを取り扱うことができます。
ちなみ、上の図だとSchema RegistryがSchemaを保持しているような概念図になっていますが、実はSchema Registryは裏ではKafka上のTopicをSchema情報の保存先として使用しているようです。
参考:Schema Registry Overview - Kafka Backend
Kafka is used as Schema Registry storage backend. The special Kafka topic (default _schemas), with a single partition, is used as a highly available write ahead log. All schemas, subject/version and ID metadata, and compatibility settings are appended as messages to this log. A Schema Registry instance therefore both produces and consumes messages under the _schemas topic.
Avro Schemaについて
Avroというのはデータフォーマットについての規約でAvro Schemaでデータの型が定義できるようです。JSON と JSON Schemaの関係と同じようにAvro と Avro Schemaの関係性を捉えるのがよいと思います。
(Avro SchemaはJSON形式で記述されるのでちょっとややこしいですが、AvroデータフォーマットとJSONデータフォーマットは別モノです。)
Schema Registryで使用できるデータタイプ(Schemaのタイプ)はいくつかあり、そのうちの1つとしてAvroがあります。Avroについては Producer, Consumer用にそれぞれ kafka-avro-console-producer、kafka-avro-console-consumerというコマンドが提供されているので、これらを使用するとシェルから簡易的にメッセージ送受信の確認ができます。また、これらのコマンドはAvro<=>JSON変換をしてくれるので、実際にデータの投入や確認を行う際はJSON形式で扱うことができます(内部的にはAvroに変換される)。
kafka-avro-console-producer --help
This tool helps to read data from standard input and publish it to Kafka.
Option Description
------ -----------
--batch-size <Integer: size> Number of messages to send in a single
batch if they are not being sent
synchronously. (default: 200)
--bootstrap-server <String: server to REQUIRED unless --broker-list
connect to> (deprecated) is specified. The server
(s) to connect to. The broker list
string in the form HOST1:PORT1,HOST2:
PORT2.
--broker-list <String: broker-list> DEPRECATED, use --bootstrap-server
instead; ignored if --bootstrap-
server is specified. The broker
list string in the form HOST1:PORT1,
HOST2:PORT2.
--compression-codec [String: The compression codec: either 'none',
compression-codec] 'gzip', 'snappy', 'lz4', or 'zstd'.
If specified without value, then it
defaults to 'gzip'
--help Print usage information.
--line-reader <String: reader_class> The class name of the class to use for
reading lines from standard in. By
default each line is read as a
separate message. (default: kafka.
tools.
ConsoleProducer$LineMessageReader)
--max-block-ms <Long: max block on The max time that the producer will
send> block for during a send request
(default: 60000)
--max-memory-bytes <Long: total memory The total memory used by the producer
in bytes> to buffer records waiting to be sent
to the server. (default: 33554432)
--max-partition-memory-bytes <Long: The buffer size allocated for a
memory in bytes per partition> partition. When records are received
which are smaller than this size the
producer will attempt to
optimistically group them together
until this size is reached.
(default: 16384)
--message-send-max-retries <Integer> Brokers can fail receiving the message
for multiple reasons, and being
unavailable transiently is just one
of them. This property specifies the
number of retries before the
producer give up and drop this
message. (default: 3)
--metadata-expiry-ms <Long: metadata The period of time in milliseconds
expiration interval> after which we force a refresh of
metadata even if we haven't seen any
leadership changes. (default: 300000)
--producer-property <String: A mechanism to pass user-defined
producer_prop> properties in the form key=value to
the producer.
--producer.config <String: config file> Producer config properties file. Note
that [producer-property] takes
precedence over this config.
--property <String: prop> A mechanism to pass user-defined
properties in the form key=value to
the message reader. This allows
custom configuration for a user-
defined message reader. Default
properties include:
parse.key=true|false
key.separator=<key.separator>
ignore.error=true|false
--request-required-acks <String: The required acks of the producer
request required acks> requests (default: 1)
--request-timeout-ms <Integer: request The ack timeout of the producer
timeout ms> requests. Value must be non-negative
and non-zero (default: 1500)
--retry-backoff-ms <Integer> Before each retry, the producer
refreshes the metadata of relevant
topics. Since leader election takes
a bit of time, this property
specifies the amount of time that
the producer waits before refreshing
the metadata. (default: 100)
--socket-buffer-size <Integer: size> The size of the tcp RECV size.
(default: 102400)
--sync If set message send requests to the
brokers are synchronously, one at a
time as they arrive.
--timeout <Integer: timeout_ms> If set and the producer is running in
asynchronous mode, this gives the
maximum amount of time a message
will queue awaiting sufficient batch
size. The value is given in ms.
(default: 1000)
--topic <String: topic> REQUIRED: The topic id to produce
messages to.
--version Display Kafka version.
kafka-avro-console-consumer --help
This tool helps to read data from Kafka topics and outputs it to standard output.
Option Description
------ -----------
--bootstrap-server <String: server to REQUIRED: The server(s) to connect to.
connect to>
--consumer-property <String: A mechanism to pass user-defined
consumer_prop> properties in the form key=value to
the consumer.
--consumer.config <String: config file> Consumer config properties file. Note
that [consumer-property] takes
precedence over this config.
--enable-systest-events Log lifecycle events of the consumer
in addition to logging consumed
messages. (This is specific for
system tests.)
--formatter <String: class> The name of a class to use for
formatting kafka messages for
display. (default: kafka.tools.
DefaultMessageFormatter)
--from-beginning If the consumer does not already have
an established offset to consume
from, start with the earliest
message present in the log rather
than the latest message.
--group <String: consumer group id> The consumer group id of the consumer.
--help Print usage information.
--isolation-level <String> Set to read_committed in order to
filter out transactional messages
which are not committed. Set to
read_uncommitted to read all
messages. (default: read_uncommitted)
--key-deserializer <String:
deserializer for key>
--max-messages <Integer: num_messages> The maximum number of messages to
consume before exiting. If not set,
consumption is continual.
--offset <String: consume offset> The offset id to consume from (a non-
negative number), or 'earliest'
which means from beginning, or
'latest' which means from end
(default: latest)
--partition <Integer: partition> The partition to consume from.
Consumption starts from the end of
the partition unless '--offset' is
specified.
--property <String: prop> The properties to initialize the
message formatter. Default
properties include:
print.timestamp=true|false
print.key=true|false
print.offset=true|false
print.partition=true|false
print.headers=true|false
print.value=true|false
key.separator=<key.separator>
line.separator=<line.separator>
headers.separator=<line.separator>
null.literal=<null.literal>
key.deserializer=<key.deserializer>
value.deserializer=<value.
deserializer>
header.deserializer=<header.
deserializer>
Users can also pass in customized
properties for their formatter; more
specifically, users can pass in
properties keyed with 'key.
deserializer.', 'value.
deserializer.' and 'headers.
deserializer.' prefixes to configure
their deserializers.
--skip-message-on-error If there is an error when processing a
message, skip it instead of halt.
--timeout-ms <Integer: timeout_ms> If specified, exit if no message is
available for consumption for the
specified interval.
--topic <String: topic> The topic id to consume on.
--value-deserializer <String:
deserializer for values>
--version Display Kafka version.
--whitelist <String: whitelist> Regular expression specifying
whitelist of topics to include for
consumption.
また、Schema Registryは管理用のREST APIが提供されているのでcurlなどを使用してアクセスすることができるようです。
Schema Registry API Reference
操作例
以下に示されている例に倣って試してみます。
Avro Schema Serializer and Deserializer - Test Drive Avro Schema
スキーマ・タイプの確認
REST APIを使ってSchema Registryに登録されているスキーマ・タイプを確認します。
[root@test12 ~]# curl http://localhost:8081/schemas/types
["JSON","PROTOBUF","AVRO"]
AVROが扱えることが確認できます。
(デフォルトではSchema Registryに上の3つのタイプのプラグインが導入済みのようです。)
Schema指定でのメッセージ送受信例1 / シンプルケース
シンプルなSchemaを使用してメッセージを送受信してみます。
{
"type": "record",
"name": "myrecord",
"fields": [
{
"name": "f1",
"type": "string"
}
]
}
上の例はメッセージ全体はrecord型(Complex typeの一種)として定義されています。"fileds"部分に具体的なフィールドの定義がされます。上の例は"f1"というString型の項目1つだけのメッセージを意味していま。
このschemaに従ったメッセージを t1-a というTopicに対して送信するProducerを起動するには、以下のコマンドを実行します。(Topicが作成されていなければ暗黙的に作成してくれるようです。)
kafka-avro-console-producer --broker-list localhost:9092 \
--property schema.registry.url=http://localhost:8081 \
--topic t1-a \
--property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}'
kafka-console-producerコマンドと違ってプロンプトが出ないので、続けて送信したいメッセージを入力してEnterを押します。
{"f1": "value1-a"}
Ctrl+Cで抜けます。
次にConsumerを起動してみます。以下のようにTopic名とSchema Registryを指定します。
kafka-avro-console-consumer --bootstrap-server localhost:9092 \
--from-beginning --topic t1-a \
--property schema.registry.url=http://localhost:8081
すると、以下のメッセージが返されます。
{"f1":"value1-a"}
送信したメッセージがそのまま返されているのであまり面白みが無いですが、裏ではシリアライズ/デシリアライズが行われているはずです。
Consumer実行時に、kafka-avro-console-consumerコマンドで--property print.schema.ids=trueを指定してみます。
[root@test12 /opt/confluent-6.2.0/bin]# ./kafka-avro-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic t1-a --property schema.registry.url=http://localhost:8081 --property print.schema.ids=true
{"f1":"value1-a"} 1
上のメッセージはSchema idとして1が紐づいていることが分かります。
curlでSchema Registryからschema id: 1 の情報を取得してみると、以下のように返されました。
[root@test12 ~]# curl http://localhost:8081/schemas/ids/1
{"schema":"{\"type\":\"record\",\"name\":\"myrecord\",\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}"}
ちなみに、このTopicをkafka-console-consumerコマンドでデシリアライズせずに見てみるとどうなるかという言うと...
[root@test12 /opt/confluent-6.2.0/bin]# ./kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic t1-a
value1-a
こうなりました。
Schema指定でのメッセージ送受信例2 / キー指定
KafkaのメッセージにはKeyを持たせることもできます。KeyとValueそれぞれSchemaを使用したメッセージを送受信してみます。
Key用のスキーマ
{"type":"string"}
Value用のスキーマ(先の例と同じ)
{
"type": "record",
"name": "myrecord",
"fields": [
{
"name": "f1",
"type": "string"
}
]
}
上のスキーマに従ったKey, Valueを持つメッセージをt2-aというTopicに送信するProducerを起動するには、以下のコマンドを実行します。
kafka-avro-console-producer --broker-list localhost:9092 \
--topic t2-a \
--property parse.key=true \
--property "key.separator= "\
--property key.schema='{"type":"string"}' \
--property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}' \
--property schema.registry.url=http://localhost:8081
送信メッセージとして以下を入力
"key1" {"f1": "value2-a"}
"AAAAA" {"f1": "TTTTT"}
"TEST01" {"f1": "XXXXXX"}
別シェルでConsumerを実行します。
kafka-avro-console-consumer --from-beginning --topic t2-a \
--bootstrap-server localhost:9092 \
--property print.key=true \
--property print.schema.ids=true \
--property schema.id.separator=: \
--property schema.registry.url=http://localhost:8081
すると、以下の結果が返されます。
"key1":2 {"f1":"value2-a"}:1
"AAAAA":2 {"f1":"TTTTT"}:1
"TEST01":2 {"f1":"XXXXXX"}:1
curlでSchema id: 2の情報を見てみます。
[root@test12 ~]# curl http://localhost:8081/schemas/ids/2
{"schema":"\"string\""}
ちなみにこちらもkafka-console-consumerコマンドでデシリアライズせずに見てみると、こんな感じになりました。
[root@test12 /opt/confluent-6.2.0/bin]# ./kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic t2-a
value2-a
TTTTT
XXXXXX
Schema指定でのメッセージ送受信例3 / 複雑なSchema
もう少し複雑なスキーマを使用してみます。
Avro Schemaの定義方法はこちらを参考に。
Apache Avro™ 1.10.2 Specification
Key用のスキーマ
{"type":"string"}
Value用のスキーマ
{
"type": "record",
"name": "userInfo",
"fields" : [
{"name": "username", "type": "string"},
{"name": "age", "type": "int"},
{"name": "phone", "type": "string"},
{"name": "address", "type": {
"type": "record",
"name" : "user_address",
"fields" : [
{"name": "street", "type": "string"},
{"name": "city", "type": "string"},
{"name": "country", "type": "string"},
{"name": "zip", "type": "string"}
]
}
}
]
}
上のスキーマを使って`test3'というTopicに対してメッセージ送受信を行います。
スキーマが長いので、上のように整形した状態のSchemaをavroSchema3.jsonというファイルで作成しておいて、それをjqコマンドで1行に変換して環境変数にセットしておきます。
[root@test12 ~/Kafka]# avroSchema3=$(jq --compact-output . avroSchema3.json)
[root@test12 ~/Kafka]# echo ${avroSchema3}
{"type":"record","name":"userInfo","fields":[{"name":"username","type":"string"},{"name":"age","type":"int"},{"name":"phone","type":"string"},{"name":"address","type":{"type":"record","name":"user_address","fields":[{"name":"street","type":"string"},{"name":"city","type":"string"},{"name":"country","type":"string"},{"name":"zip","type":"string"}]}}]}
以下コマンドでProducerを起動します。
kafka-avro-console-producer --broker-list localhost:9092 --topic test3 \
--property parse.key=true \
--property "key.separator= "\
--property key.schema='{"type":"string"}' \
--property value.schema=${avroSchema3} \
--property schema.registry.url=http://localhost:8081
別シェルでConsumerを起動します。
kafka-avro-console-consumer --from-beginning --topic test3 \
--bootstrap-server localhost:9092 \
--property print.key=true \
--property print.schema.ids=true \
--property schema.id.separator=: \
--property schema.registry.url=http://localhost:8081
以下のProducerから以下のデータを送信してみます。
"A001" {"username": "Yamada Ichiro", "age": 31, "phone": "1234-5678", "address":{"street": "1-1", "city": "chiba", "country": "Japan", "zip": "111-2222"}}
"A002" {"username": "Tanaka Jiro", "age": 32, "phone": "2222-2222", "address":{"street": "2-2", "city": "tokyo", "country": "Japan", "zip": "111-2222"}}
Consumerでは以下のデータが受信されました。
"A001":2 {"username":"Yamada Ichiro","age":31,"phone":"1234-5678","address":{"street":"1-1","city":"chiba","country":"Japan","zip":"111-2222"}}:3
"A002":2 {"username":"Tanaka Jiro","age":32,"phone":"2222-2222","address":{"street":"2-2","city":"tokyo","country":"Japan","zip":"111-2222"}}:3
kafka-console-consumerコマンドを使ってデシリアライズを行わずにTopictest3のメッセージを受信してみるとこんな感じになりました。
[root@test12 /opt/confluent-6.2.0/bin]# ./kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic test3
Yamada Ichiro>1234-56781-1
chiba
Japan111-2222
Tanaka Jiro@2222-22222-2
tokyo
Japan111-2222
恐らくAvroはバイナリでデータ保持していると思われ、改行は正しく表示されていないのではないかと思います。これで、内部的にはきちんとSchemaが使われてシリアライズ/デシリアライズが行われているであろうということが分かります。
ちなみに、Avro用のProducer/Consumerに戻ってSchemaでの定義に対してフィールドが足りないデータを投入するとProducerでエラーになりました。
送信
"A003" {"username": "Suzuki Saburo"}
Producerがエラーで終了
org.apache.kafka.common.errors.SerializationException: Error deserializing json {"username": "Suzuki Saburo"} to Avro of schema {"type":"record","name":"userInfo","fields":[{"name":"username","type":"string"},{"name":"age","type":"int"},{"name":"phone","type":"string"},{"name":"address","type":{"type":"record","name":"user_address","fields":[{"name":"street","type":"string"},{"name":"city","type":"string"},{"name":"country","type":"string"},{"name":"zip","type":"string"}]}}]}
Caused by: org.apache.avro.AvroTypeException: Expected int. Got END_OBJECT
at org.apache.avro.io.JsonDecoder.error(JsonDecoder.java:511)
at org.apache.avro.io.JsonDecoder.readInt(JsonDecoder.java:165)
at org.apache.avro.io.ValidatingDecoder.readInt(ValidatingDecoder.java:82)
at org.apache.avro.generic.GenericDatumReader.readInt(GenericDatumReader.java:551)
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:195)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160)
at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:259)
at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:247)
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:160)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)
at io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils.toObject(AvroSchemaUtils.java:190)
at io.confluent.kafka.formatter.AvroMessageReader.readFrom(AvroMessageReader.java:121)
at io.confluent.kafka.formatter.SchemaMessageReader.readMessage(SchemaMessageReader.java:344)
at kafka.tools.ConsoleProducer$.main(ConsoleProducer.scala:51)
at kafka.tools.ConsoleProducer.main(ConsoleProducer.scala)
スキーマ情報の整形
先にも示したように、スキーマの情報はSchema RegistryからRESTで取得することができます。
例えば、以下のようにスキーマ情報を取得したとします。(Schema id: 5 の情報をGETしている例)
[root@test12 ~]# curl http://localhost:8081/schemas/ids/5
{"schema":"{\"type\":\"record\",\"name\":\"FILEA\",\"namespace\":\"value.SOURCEDB.VSAM.TEST01\",\"fields\":[{\"name\":\"STAT\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"STAT\",\"length\":1},\"null\"],\"doc\":\"\",\"default\":\"\"},{\"name\":\"NUMB\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"NUMB\",\"length\":6},\"null\"],\"doc\":\"\",\"default\":\"\"},{\"name\":\"NAME\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"NAME\",\"length\":20},\"null\"],\"doc\":\"\",\"default\":\"\"},{\"name\":\"ADDRX\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"ADDRX\",\"length\":20},\"null\"],\"doc\":\"\",\"default\":\"\"},{\"name\":\"PHONE\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"PHONE\",\"length\":8},\"null\"],\"doc\":\"\",\"default\":\"\"},{\"name\":\"DATEX\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"DATEX\",\"length\":8},\"null\"],\"doc\":\"\",\"default\":\"\"},{\"name\":\"AMOUNT\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"AMOUNT\",\"length\":8},\"null\"],\"doc\":\"\",\"default\":\"\"},{\"name\":\"COMMENT\",\"type\":[{\"type\":\"string\",\"logicalType\":\"CHARACTER\",\"dbColumnName\":\"COMMENT\",\"length\":9},\"null\"],\"doc\":\"\",\"default\":\"\"}]}"}
このように、結果は JSONフォーマットで"schema" という要素に1行の文字列でスキーマ情報が返されます。こんな感じで=> {"schema":"xxx"}。 このxxx部分がJSON形式で表された求めるスキーマなのですが文字列としてJSONデータが1要素の値として返されるので、ダブルクォーテーションがエスケープされています(バックスラッシュ付き)。この例のように少し複雑なスキーマが返されると非常に見にくいですね。
そこで、以下の様にコマンドを発行すると、スキーマ情報だけをきれいに整形して出力することができます。
[root@test12 ~]# curl -s http://localhost:8081/schemas/ids/5 | jq -r .schema | jq .
{
"type": "record",
"name": "FILEA",
"namespace": "value.SOURCEDB.VSAM.TEST01",
"fields": [
{
"name": "STAT",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "STAT",
"length": 1
},
"null"
],
"doc": "",
"default": ""
},
{
"name": "NUMB",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "NUMB",
"length": 6
},
"null"
],
"doc": "",
"default": ""
},
{
"name": "NAME",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "NAME",
"length": 20
},
"null"
],
"doc": "",
"default": ""
},
{
"name": "ADDRX",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "ADDRX",
"length": 20
},
"null"
],
"doc": "",
"default": ""
},
{
"name": "PHONE",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "PHONE",
"length": 8
},
"null"
],
"doc": "",
"default": ""
},
{
"name": "DATEX",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "DATEX",
"length": 8
},
"null"
],
"doc": "",
"default": ""
},
{
"name": "AMOUNT",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "AMOUNT",
"length": 8
},
"null"
],
"doc": "",
"default": ""
},
{
"name": "COMMENT",
"type": [
{
"type": "string",
"logicalType": "CHARACTER",
"dbColumnName": "COMMENT",
"length": 9
},
"null"
],
"doc": "",
"default": ""
}
]
}
簡単に上のコマンド補足しておきます。
curlで取ってきた情報をjqコマンドで必要な部分を抽出しています。やっているのは以下の通りです。
- curlで情報取得
- jqでschema要素の値のみ抽出
- schema要素の値を改めてjqで整形して表示
参考
スクリプト
#!/bin/bash
sudo /opt/confluent-6.2.0/bin/kafka-topics --list --bootstrap-server localhost:9092
#!/bin/bash
if [ $# -lt 1 ] ; then
echo "Usage: "
echo " " $0 " <topicName>"
exit 0
fi
topicName=$1
sudo /opt/confluent-6.2.0/bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic ${topicName} --property schema.registry.url=http://localhost:8081
#sudo /opt/confluent-6.2.0/bin/kafka-avro-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic ${topicName} --property schema.registry.url=http://localhost:8081 --property print.schema.ids=true
#!/bin/bash
if [ $# -lt 1 ] ; then
echo "Usage: "
echo " " $0 " <schemaID>"
exit 0
fi
schemaID=$1
curl --silent http://localhost:8081/schemas/ids/${schemaID} | jq -r .schema | jq .