Cannot parse <null> schema

[robshep]

Describe the bug

Reconnecting an updated consumer application that had NOT changed any of the following:

• Any connection or topic details
• Any message object details used to generate the AvroSchema

This system was restarted from an update of application code unrelated to pulsar handling, and has two consumers.
At first both consumers were failing but after a restart one is OK and the other continues to fail with the error:

org.apache.pulsar.shade.org.apache.avro.SchemaParseException: Cannot parse <null> schema

I could see the schemas using pulsar-admin, and they look exactly as they should, and the producers are tied to the same.

For the two consumar connections in this application they should have been binding to schema versions 1 & 2 respectively, but both had the error mentioning schema version 0

My connection code:

consumerIngressV1 = pulsarClient.newConsumer(AvroSchema.of(IngressMessageV1.class))
    .subscriptionName(subscriptionName)
    .topic(topics)
    .subscribe();

The stacktrace:

2020-10-27 22:48:12,322 [pool-3-thread-8] INFO o.a.p.client.impl.schema.AvroSchema - Load schema reader for version(0), schema is : , schemaInfo: {
 "name": "",
 "schema": "",
 "type": "NONE",
 "properties": {}
} -- 
2020-10-27 22:48:12,323 [pool-3-thread-8] ERROR io.pethau.pl.PipelineService - Exception: org.apache.pulsar.shade.org.apache.avro.SchemaParseException: Cannot parse <null> schema - {} - {} - {} -- 
org.apache.pulsar.shade.com.google.common.util.concurrent.UncheckedExecutionException: org.apache.pulsar.shade.org.apache.avro.SchemaParseException: Cannot parse <null> schema
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2050)
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache.get(LocalCache.java:3951)
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3973)
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4957)
    at org.apache.pulsar.client.impl.schema.StructSchema.decode(StructSchema.java:107)
    at org.apache.pulsar.client.impl.MessageImpl.getValue(MessageImpl.java:293)
    at io.pethau.pl.PipelineService.lambda$run$2(PipelineService.java:103)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
    at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.pulsar.shade.org.apache.avro.SchemaParseException: Cannot parse <null> schema
    at org.apache.pulsar.shade.org.apache.avro.Schema.parse(Schema.java:1597)
    at org.apache.pulsar.shade.org.apache.avro.Schema$Parser.parse(Schema.java:1396)
    at org.apache.pulsar.shade.org.apache.avro.Schema$Parser.parse(Schema.java:1384)
    at org.apache.pulsar.client.impl.schema.StructSchema.parseAvroSchema(StructSchema.java:182)
    at org.apache.pulsar.client.impl.schema.AvroSchema.loadReader(AvroSchema.java:99)
    at org.apache.pulsar.client.impl.schema.StructSchema$1.load(StructSchema.java:75)
    at org.apache.pulsar.client.impl.schema.StructSchema$1.load(StructSchema.java:72)
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3527)
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2276)
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2154)
    at org.apache.pulsar.shade.com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2044)
    ... 11 common frames omitted

To Reproduce
Unable to re-produce.

Expected behaviour
Consumers connect to the topics using the latest version from before the restart.

Desktop (please complete the following information):

• OS: Docker, standalone, v2.6.0

Additional context
None.

[robshep]

A log file showing the same issue encountered just now.

A restart of the application recovered, but seemingly wasn't able to recover itself.

thingping.log

[codelipenghui]

@robshep Could you please provide more information about the schema configuration? Which schema compatibility policy are you using? how the schema of the topic looks like now(You can use the pulsar-admin schemas to get the current schema of the topic)

[robshep]

Thank you,

I'm afraid I don't know how to get the schema compatibility policy of a topic - please can you advise. I don't see a cli command to show this)

On the few occasions this has happened, it typically works OK after restarting the application.
On once occasion it has failed to recover so I used a fresh broker cluster. (in testing environment so no data lost)

You can see in the original post the NULL schema that is returned as a symptom of this problem,
whereas below is a proper one as logged upon connection.

  "name": "myorg/myapp/ingress",
  "schema": {
    "type": "record",
    "name": "IngressMessageV1",
    "namespace": "io.myapp.shared.pl",
    "fields": [
      {
        "name": "attributes",
        "type": [
          "null",
          {
            "type": "map",
            "values": "string"
          }
        ]
      },
      {
        "name": "timestamp",
        "type": "long"
      },
      {
        "name": "rawMessage",
        "type": [
          "null",
          {
            "type": "bytes",
            "java-class": "[B"
          }
        ]
      }
    ]
  },
  "type": "AVRO",
  "properties": {
    "__alwaysAllowNull": "true",
    "__jsr310ConversionEnabled": "false"
  }
}```

[codelipenghui]

@robshep Thanks, seems the schema definition is not the problem, I think maybe the problem is the reader cache in the StructSchema.
@congbobo184 Could you please help take a look at this issue?

[congbobo184]

@robshep hello, I look up your log and are you consume more than one topic in your consumer? it look like one topic does not have schema.

[robshep]

Hi, thanks for taking a look.
There are 3 topics in use in this deployment.
All have schemas.
None of this code, or the Payload objects have changed for many months.

I notice in App1 there is another topic -RETRY created presumably to handle redelivery. I don't maintain a schema here.

The log above is from restarting App1 below but I have seen the same issue with the "Null Schema" also when restarting App3 also.

App1:
1 x consumer, 1 x producer

@Bean(destroyMethod = "close")
@Autowired
public Consumer<UpdateMessageV1> updateListener(PulsarUpdateListener pulsarUpdateListener) throws PulsarClientException {
        return pulsarClient().newConsumer(AvroSchema.of(UpdateMessageV1.class))
                .messageListener(pulsarUpdateListener)
                .subscriptionMode(SubscriptionMode.Durable)
                .subscriptionName("app-tp-updater")
                .negativeAckRedeliveryDelay(20, TimeUnit.SECONDS) // try again after 20 seconds
                .topic("persistent://myorg/myappB/update")
                .enableRetry(true)
                .subscriptionType(SubscriptionType.Shared)
                .subscribe();
    }

    @Bean(destroyMethod = "close")
    public Producer notifyExpiryProducer() throws PulsarClientException {
        return pulsarClient().newProducer(AvroSchema.of(ExpiryNotificationV1.class))
                .topic("persistent://myorg/myappB/notify:expiry")
                .create();
    }

App 2:
1 x producer

final Producer<IngressMessageV1> producer = pulsarClient.newProducer(AvroSchema.of(IngressMessageV1.class))
    .topic("persistent://myorg/myappA/ingress")
    .create();

App3:
1 x producer, 2 x consumer

producer = pulsarClient.newProducer(AvroSchema.of(UpdateMessageV1.class))
                        .producerName("myapp-tp-update-consumer")
                        .topic("persistent://myorg/myappB/update")
                        .create();

consumerTPNotifsV1 = pulsarClient.newConsumer(AvroSchema.of(ExpiryNotificationV1.class))
                .subscriptionName("app-tp-expiry")
                .topic( "persistent://myorg/appB/notify:expiry")
                .subscribe();

consumerIngressV1 = pulsarClient.newConsumer(AvroSchema.of(IngressMessageV1.class))
                .subscriptionName("myapp-ingress")
                .topic("persistent://myorg/myappA/ingress")
                .subscribe();

Thanks

Rob

[congbobo184]

@robshep could you please provide more detailed information from app3, log and which topic the consumer Cannot parse <null> schema

[robshep]

Thank you,

Attached is the application log from the first time this happened.

It isn't clear exaclty which topic the consumer finds a null schema, but you can see in this example it was both consumers.

On this occasion a plain restart recovered one of them (persistent://myorg/appB/notify:expiry) but the other (persistent://myorg/myappA/ingress) did not recover (same error of Null schema on each message received) and so I just trashed the standalone pulsar and started a new instance.

It has done it since, but a restart recovered them.

Thanks again.

main.log.2020-10-27.txt

[congbobo184]

if restart, this topic persistent://myorg/myappA/ingress not recover you can use admin api to check is the schema exist.

[robshep]

Hi,

This is still happening and we still need to keep restarting connected producers/consumers after updating our application.
Occasionally if 20+restarts does not clear it, we destroy the (docker standalone) broker and re-build.

• I'm afraid we may not validate Pulsar into production and will have to fall back to kafka 😞

@congbobo184 All the expected topics appear using pulsar-admin at any stage in the process: before, during a failure, after etc.

[congbobo184]

@robshep
Sorry to hear that,

AvroSchema has been in the production environment for a long time. It is a stable feature.

I think you can use one topic to test and use ./pulsar-admin schemas get persistent://tenant/namespace/topic --version to find out this version of schema in broker or provide a stable way to reproduce for me, I will reproduce it. include. (schema compatibility, all version schema, producer and consumer update order and update schema.)

I think this is not a problem that can prevent you from entering the production environment, we can try to solve this problem together. You can ping me in pulsar slack general channel. my clack name is CongBo. Expect you to find me :)

[robshep]

Thank you, If the components are generally stable then there could be issues with my setup? In this case I am grateful for your support, and will look you up on slack With thanks R Get Outlook for iOS<https://aka.ms/o0ukef>

…

________________________________ From: congbo <notifications@github.com> Sent: Monday, January 18, 2021 2:23:49 AM To: apache/pulsar <pulsar@noreply.github.com> Cc: Rob Shepherd <rs@datacymru.net>; Mention <mention@noreply.github.com> Subject: Re: [apache/pulsar] Cannot parse <null> schema (#8429) @robshep<https://github.com/robshep> Sorry to hear that, AvroSchema has been in the production environment for a long time. It is a stable feature. I think you can use one topic to test and use ./pulsar-admin schemas get persistent://tenant/namespace/topic --version to find out this version of schema in broker or provide a stable way to reproduce for me, I will reproduce it. include. (schema compatibility, all version schema, producer and consumer update order and update schema.) I think this is not a problem that can prevent you from entering the production environment, we can try to solve this problem together. You can ping me in pulsar slack general channel. my clack name is CongBo. Expect you to find me 🔢 — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub<#8429 (comment)>, or unsubscribe<https://github.com/notifications/unsubscribe-auth/AANXMU4WPGB3KXWTSILZTPTS2OLTLANCNFSM4THYHL3Q>.