[Go Functions] publishfunc example is broken

[flowchartsman]

Describe the bug
If you attempt to run the publishfunc example (dispatching to alternate topics), your function will panic.

building/creating:

$ GOOS=linux GOARCH=amd64 go build -o publishfunc
$ pulsarctl functions create --tenant public --namespace default --name publishfunc --inputs persistent://public/default/test --go `pwd`/publishfunc

pulsar log:

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x48 pc=0xb0d63b]

goroutine 1 [running]:
github.com/apache/pulsar/pulsar-function-go/pf.(*goInstance).getProducer(0xc000346000, 0xc8ce41, 0xd, 0xc00017c2a0, 0x6c, 0xe0, 
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/instance.go:225 
github.com/apache/pulsar/pulsar-function-go/pf.newGoInstance.func1(0xc8ce41, 0xd, 0x6b, 0x6b)
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/instance.go:71 
github.com/apache/pulsar/pulsar-function-go/pf.(*FunctionContext).NewOutputMessage(...)
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/context.go:158
main.PublishFunc(0xd8ba60, 0xc0002b84b0, 0xc00039200a, 0x6b, 0x6b, 0x0, 0x0)
      /Users/me/myproj/functions/publishfunc/main.go:40 +0xd5
reflect.Value.call(0xb96160, 0xcc5fe8, 0x13, 0xc85522, 0x4, 0xc0002b84e0, 0x2, 0x2, 0x2, 0x18, ...)
      /Users/me/opt/go/src/reflect/value.go:476 +0x8c7
reflect.Value.Call(0xb96160, 0xcc5fe8, 0x13, 0xc0002b84e0, 0x2, 0x2, 0x1, 0x2, 0x203000)
      /Users/me/opt/go/src/reflect/value.go:337 +0xb9
github.com/apache/pulsar/pulsar-function-go/pf.newFunction.func1(0xd8ba60, 0xc0002b84b0, 0xc00039200a, 0x6b, 0x6b, 0xc24120, 002b84b0, 0xc0001b1cb0, 0x49a426)
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/function.go:126 
github.com/apache/pulsar/pulsar-function-go/pf.pulsarFunction.process(0xc000308000, 0xd8ba60, 0xc0002b84b0, 0xc00039200a, 0x6b, c0002b84b0, 0x1e23a578, 0xa1c3b3e10c47, ...)
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/function.go:47 
github.com/apache/pulsar/pulsar-function-go/pf.(*goInstance).handlerMsg(0xc000346000, 0xd93460, 0xc000306750, 0x0, 0x0, 0x0, 0x0, 
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/instance.go:340 
github.com/apache/pulsar/pulsar-function-go/pf.(*goInstance).startFunction(0xc000346000, 0xd7cda0, 0xc000308000, 0x0, 0x0)
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/instance.go:165 
github.com/apache/pulsar/pulsar-function-go/pf.Start(0xb96160, 0xcc5fe8)
      /Users/me/go/pkg/mod/github.com/apache/pulsar/pulsar-function-go@v0.0.0-20201214050154-ff3dbce84d47/pf/function.go:171 
main.main()
      /Users/me/myproj/functions/publishfunc/main.go:53 +0x39

To Reproduce
Compile the example exactly as it appears.

Expected behavior
"publish-topic" topic appears containing messages

[flowchartsman]

This corresponds to the line in pf/instance.go:

batchBuilder := gi.context.instanceConf.funcDetails.Sink.ProducerSpec.BatchBuilder

In my testing, it is ProducerSpec which is nil

[flowchartsman]

@wolfstudy @sijie I believe it was just an issue of a missing nil check for non-sink functions, since the purpose of the code in the first place seems to check for a configured BatchBuilder and then setting the new Producer's batch builder to use it. That said, now that I look at it more closely, I'm relatively concerned with the number of log messages indicating a new producer server-side for each message. Is this okay and, if so, is there any way we can suppress these? Seems like potentially a lot of memory pressure to be allocating new producers for the same topic for every message just to throw them away, though.

01:54:38.326 [ForkJoinPool.commonPool-worker-3] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] Created new producer: Producer{topic=PersistentTopic{topic=persistent://public/default/test}, client=/127.0.0.1:35876, producerName=standalone-0-3743, producerId=874}
01:54:39.335 [pulsar-io-50-8] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876][persistent://public/default/test] Creating producer. producerId=875
01:54:39.336 [ForkJoinPool.commonPool-worker-2] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] persistent://public/default/test configured with schema false
01:54:39.336 [ForkJoinPool.commonPool-worker-2] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] Created new producer: Producer{topic=PersistentTopic{topic=persistent://public/default/test}, client=/127.0.0.1:35876, producerName=standalone-0-3744, producerId=875}
01:54:40.300 [pulsar-io-50-8] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876][persistent://public/default/test] Creating producer. producerId=876
01:54:40.301 [ForkJoinPool.commonPool-worker-6] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] persistent://public/default/test configured with schema false
01:54:40.301 [ForkJoinPool.commonPool-worker-6] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] Created new producer: Producer{topic=PersistentTopic{topic=persistent://public/default/test}, client=/127.0.0.1:35876, producerName=standalone-0-3745, producerId=876}
01:54:41.300 [pulsar-io-50-8] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876][persistent://public/default/test] Creating producer. producerId=877
01:54:41.301 [ForkJoinPool.commonPool-worker-3] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] persistent://public/default/test configured with schema false
01:54:41.301 [ForkJoinPool.commonPool-worker-3] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] Created new producer: Producer{topic=PersistentTopic{topic=persistent://public/default/test}, client=/127.0.0.1:35876, producerName=standalone-0-3746, producerId=877}
01:54:42.300 [pulsar-io-50-8] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876][persistent://public/default/test] Creating producer. producerId=878
01:54:42.301 [ForkJoinPool.commonPool-worker-6] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] persistent://public/default/test configured with schema false
01:54:42.301 [ForkJoinPool.commonPool-worker-6] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876] Created new producer: Producer{topic=PersistentTopic{topic=persistent://public/default/test}, client=/127.0.0.1:35876, producerName=standalone-0-3747, producerId=878}
01:54:43.300 [pulsar-io-50-8] INFO  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:35876][persistent://public/default/test] Creating producer. producerId=879