This message was deleted pulsar #general

Join Slack

This message was deleted.

# general

Slackbot

04/15/2023, 10:08 PM

This message was deleted.

Enrico Olivelli

04/16/2023, 1:46 PM

I suggest to set it using code, then you can get it with pulsar-admin in order to learn the syntax

Anthony Ikeda

04/16/2023, 6:22 PM

So I've done that but the schema is coming back as 2 different SchemaInfo's, one for the key and one for the value:

Copy code

MultiVersionAvroReader     : Load schema reader for version(0), schema is : {"type":"record","name":"ProductKey","namespace":"com.example.pulsarproducerapp","fields":[{"name":"dept","type":["null","string"],"default":null},{"name":"sku","type":["null","string"],"default":null}]}, schemaInfo: {
  "name": "",
  "schema": {
    "type": "record",
    "name": "ProductKey",
    "namespace": "com.example.pulsarproducerapp",
    "fields": [
      {
        "name": "dept",
        "type": [
          "null",
          "string"
        ]
      },
      {
        "name": "sku",
        "type": [
          "null",
          "string"
        ]
      }
    ]
  },
  "type": "AVRO",
  "timestamp": 0,
  "properties": {
    "__alwaysAllowNull": "true",
    "__jsr310ConversionEnabled": "false"
  }
}


MultiVersionAvroReader     : Load schema reader for version(0), schema is : {"type":"record","name":"ProductValue","namespace":"com.example.pulsarproducerapp","fields":[{"name":"createdBy","type":["null","string"],"default":null},{"name":"description","type":["null","string"],"default":null},{"name":"name","type":["null","string"],"default":null}]}, schemaInfo: {
  "name": "",
  "schema": {
    "type": "record",
    "name": "ProductValue",
    "namespace": "com.example.pulsarproducerapp",
    "fields": [
      {
        "name": "createdBy",
        "type": [
          "null",
          "string"
        ]
      },
      {
        "name": "description",
        "type": [
          "null",
          "string"
        ]
      },
      {
        "name": "name",
        "type": [
          "null",
          "string"
        ]
      }
    ]
  },
  "type": "AVRO",
  "timestamp": 0,
  "properties": {
    "__alwaysAllowNull": "true",
    "__jsr310ConversionEnabled": "false"
  }
}

Here is how I'm setting the schema in code (Java):

Copy code

Schema<ProductKey> keySchema = Schema.AVRO(ProductKey.class);
        Schema<ProductValue> valueSchema = Schema.AVRO(ProductValue.class);

        Schema<KeyValue<ProductKey, ProductValue>> schema = Schema.KeyValue(
                keySchema,
                valueSchema,
                KeyValueEncodingType.INLINE);
        Producer<KeyValue<ProductKey, ProductValue>> producer = null;

        producer = client.newProducer(schema)
                    .topic(TOPIC)
                    .create();
        ProductKey key = new ProductKey();
        key.setDept(dept);
        key.setSku(sku);

        ProductValue value = new ProductValue();
        value.setName(name);
        value.setDescription(description);
        value.setCreatedBy("me");

        producer.newMessage().value(new KeyValue<>(key, value)).send();
        producer.close();

How would this translate to

pulsar-admin schemas upload

Anthony Ikeda

04/17/2023, 4:15 PM

I did notice when using the Debezium PostgreSQL Source, the schema (key and value) were contained in the same SchemaInfo representation, which I’m trying to recreate; the main reason for going down this path.

Enrico Olivelli

04/22/2023, 3:18 PM

I am sorry for late reply

Enrico Olivelli

04/22/2023, 3:19 PM

You are showing the logs. What about getting the schema with pulsar-admin? It should be left intact

Anthony Ikeda

04/24/2023, 3:33 PM

Okay, I’ll give that a try.

3 Views

Open in Slack

Previous Next