ISSUE-5454: mysql JDBC Sink - consumer error

[sijie]

Original Issue: apache#5454

---

Describe the bug
Running mysql jdbc sink (helped in issue #5418) but unable to receive data from producer.

To Reproduce
Steps to reproduce the behavior:

1. Define Test schema and send producer data

    class Test(Record):
        id = Integer()
        name = String()
   
    producer = client.create_producer(
                    'pulsar-mysql-jdbc-sink-topic',
                    schema=AvroSchema(Test))
   

2. Receiving error in SINK logs:

19:25:55.712 [pulsar-client-io-1-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - [pulsar-mysql-jdbc-sink-topic][public/default/pulsar-mysql-jdbc-sink] Subscribing to topic on cnx [id: 0x79b6a514, L:/135.25.67.33:33184 - R:mtn21cvaas04.mt.att.com/135.25.67.33:6650]
19:25:55.734 [pulsar-client-io-1-1] INFO org.apache.pulsar.client.impl.ConsumerImpl - [pulsar-mysql-jdbc-sink-topic][public/default/pulsar-mysql-jdbc-sink] Subscribed to topic on mtn21cvaas04.mt.att.com/135.25.67.33:6650 -- consumer: 0

19:29:41.376 [pulsar-client-io-1-1] WARN com.scurrilous.circe.checksum.Crc32cIntChecksum - Failed to load Circe JNI library. Falling back to Java based CRC32c provider
19:29:41.427 [public/default/pulsar-mysql-jdbc-sink-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [public/default/pulsar-mysql-jdbc-sink:0] Uncaught exception in Java Instance
java.lang.NullPointerException: null
at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:770) ~[com.google.guava-guava-21.0.jar:?]
at com.google.common.cache.LocalCache.get(LocalCache.java:4153) ~[com.google.guava-guava-21.0.jar:?]
at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:4158) ~[com.google.guava-guava-21.0.jar:?]
at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:5147) ~[com.google.guava-guava-21.0.jar:?]
at org.apache.pulsar.client.impl.schema.StructSchema.decode(StructSchema.java:94) ~[org.apache.pulsar-pulsar-client-original-2.4.1.jar:2.4.1]
at org.apache.pulsar.client.impl.schema.AutoConsumeSchema.decode(AutoConsumeSchema.java:72) ~[org.apache.pulsar-pulsar-client-original-2.4.1.jar:2.4.1]
at org.apache.pulsar.client.impl.schema.AutoConsumeSchema.decode(AutoConsumeSchema.java:36) ~[org.apache.pulsar-pulsar-client-original-2.4.1.jar:2.4.1]
at org.apache.pulsar.client.api.Schema.decode(Schema.java:97) ~[java-instance.jar:?]
at org.apache.pulsar.client.impl.MessageImpl.getValue(MessageImpl.java:268) ~[org.apache.pulsar-pulsar-client-original-2.4.1.jar:2.4.1]
at org.apache.pulsar.functions.source.PulsarRecord.getValue(PulsarRecord.java:74) ~[org.apache.pulsar-pulsar-functions-instance-2.4.1.jar:2.4.1]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.readInput(JavaInstanceRunnable.java:463) ~[org.apache.pulsar-pulsar-functions-instance-2.4.1.jar:?]
at org.apache.pulsar.functions.instance.JavaInstanceRunnable.run(JavaInstanceRunnable.java:236) [org.apache.pulsar-pulsar-functions-instance-2.4.1.jar:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_212]
19:29:41.504 [public/default/pulsar-mysql-jdbc-sink-0] INFO org.apache.pulsar.functions.instance.JavaInstanceRunnable - Closing instance
19:29:41.510 [public/default/pulsar-mysql-jdbc-sink-0] INFO org.apache.pulsar.client.impl.ConsumerImpl - [pulsar-mysql-jdbc-sink-topic] [public/default/pulsar-mysql-jdbc-sink] Closed consumer
19:29:41.512 [public/default/pulsar-mysql-jdbc-sink-0] INFO org.apache.pulsar.io.jdbc.JdbcAbstractSink - Closed jdbc connection: jdbc:mysql://127.0.0.1:3306/pulsar_mysql_jdbc_sink?useUnicode=true&characterEncoding=utf-8&useSSL=false
19:29:41.518 [public/default/pulsar-mysql-jdbc-sink-0] INFO org.apache.pulsar.functions.instance.JavaInstanceRunnable - Unloading JAR files for function InstanceConfig(instanceId=0, functionId=5703beea-ca01-446b-8d1a-35167aeb2cac, functionVersion=df4c3460-d718-4684-9109-3d081cfb6f1e, functionDetails=tenant: "public"
namespace: "default"
name: "pulsar-mysql-jdbc-sink"
className: "org.apache.pulsar.functions.api.utils.IdentityFunction"
autoAck: true
parallelism: 1
source {
typeClassName: "org.apache.pulsar.client.api.schema.GenericRecord"
inputSpecs {
key: "pulsar-mysql-jdbc-sink-topic"
value {
}
}
cleanupSubscription: true
}
sink {
className: "org.apache.pulsar.io.jdbc.JdbcAutoSchemaSink"
configs: "{"userName":"root","password":"jdbc","jdbcUrl":"jdbc:mysql://127.0.0.1:3306/pulsar_mysql_jdbc_sink?useUnicode\u003dtrue\u0026characterEncoding\u003dutf-8\u0026useSSL\u003dfalse","tableName":"pulsar_mysql_jdbc_sink"}"
typeClassName: "org.apache.pulsar.client.api.schema.GenericRecord"
}
resources {
cpu: 1.0
ram: 1073741824
disk: 10737418240
}
componentType: SINK
, maxBufferedTuples=1024, functionAuthenticationSpec=null, port=37644, clusterName=standalone)
19:29:41.518 [main] INFO org.apache.pulsar.functions.runtime.JavaInstanceStarter - RuntimeSpawner quit, shutting down JavaInstance
19:29:41.520 [main] INFO org.apache.pulsar.client.impl.PulsarClientImpl - Client closing. URL: pulsar://mtn21cvaas04.mt.att.com:6650
19:29:41.530 [pulsar-client-io-1-1] INFO org.apache.pulsar.client.impl.ClientCnx - [id: 0x79b6a514, L:/135.25.67.33:33184 ! R:mtn21cvaas04.mt.att.com/135.25.67.33:6650]
Disconnected

Expected behavior
Expect to see no error and some record in mysql DB.

Desktop (please complete the following information):

• OS: ubuntu 16.4

Additional context
It maybe some miss-configuration and not a bug.
SINK is using sample schema provided by pulsar instructions:

{
"type": "AVRO",
"schema": "{"type":"record","name":"Test","fields":[{"name":"id","type":["null","int"]},{"name":"name","type":["null","string"]}]}",
"properties": {}
}