Issue #432 propagate disableForceTopicCreation

[flowchartsman]

fixes #432

[merlimat]

Change looks good. Can you also add a test so that we can ensure the behavior will not be broken later on?

[flowchartsman]

@merlimat Test committed. Verified working and verified that it fails without

disableForceTopicCreation:  c.disableForceTopicCreation,

[flowchartsman]

This test appears to have exposed a race condition in access to the state of the partition consumer. Output from local testing:

WARNING: DATA RACE
Read at 0x00c00039a918 by goroutine 9:
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).clearQueueAndGetNextMessage()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:911 +0x68
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).clearReceiverQueue()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:932 +0x84
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).grabConn()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:861 +0x1312
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).reconnectToBroker()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:802 +0x238
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).runEventsLoop.func2()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:711 +0xab

Previous write at 0x00c00039a918 by goroutine 70:
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).internalClose()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:772 +0x68f
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).runEventsLoop()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:732 +0x29b

Goroutine 9 (running) created at:
  github.com/apache/pulsar-client-go/pulsar.(*partitionConsumer).runEventsLoop()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:704 +0x146

Goroutine 70 (finished) created at:
  github.com/apache/pulsar-client-go/pulsar.newPartitionConsumer()
      /Users/me/pulsar-client-go/pulsar/consumer_partition.go:184 +0xead
  github.com/apache/pulsar-client-go/pulsar.(*consumer).internalTopicSubscribeToPartitions.func1()
      /Users/me/pulsar-client-go/pulsar/consumer_impl.go:314 +0x7d3

These correspond with access to pc.state, which is not protected with a mutex or accessed atomically. I will add an issue to address this.

[flowchartsman]

The issue I created (#448) looks to have a viable fix in #451, so if these tests look good, this should be good to land.

[wolfstudy]

@flowchartsman Thanks your work for this, the change LGTM +1, and can you merge master code and fix test case? Please make sure the CI is ok.

[flowchartsman]

@wolfstudy, @merlimat failing tests appear unrelated, can you confirm?

--- FAIL: TestDLQMultiTopics (0.09s)
    consumer_test.go:1088: 
        	Error Trace:	consumer_test.go:1088
        	Error:      	Expected nil, but got: unable to subscribe to topic=persistent://public/default/my-topic-262453276-9: server error: TopicNotFound: Topic does not exist
        	Test:       	TestDLQMultiTopics
panic: runtime error: invalid memory address or nil pointer dereference [recovered]
	panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0xf09629]

goroutine 2734 [running]:
testing.tRunner.func1(0xc000219100)
	/usr/local/go/src/testing/testing.go:874 +0x69f
panic(0x10d2620, 0x19ceff0)
	/usr/local/go/src/runtime/panic.go:679 +0x1b2
github.com/apache/pulsar-client-go/pulsar.TestDLQMultiTopics(0xc000219100)
	/pulsar-client-go/pulsar/consumer_test.go:1089 +0x6a9
testing.tRunner(0xc000219100, 0x11e8698)
	/usr/local/go/src/testing/testing.go:909 +0x19a
created by testing.(*T).Run
	/usr/local/go/src/testing/testing.go:960 +0x652
FAIL	github.com/apache/pulsar-client-go/pulsar	32.780s

[merlimat]

    	Error:      	Expected nil, but got: unable to subscribe to topic=persistent://public/default/my-topic-262453276-9: server error: TopicNotFound: Topic does not exist

@flowchartsman I think it is related to the change. It looks like the disableForceTopicCreation is passed in a case where is not supposed to be there, and the topic is not getting created, leading to the test failure.

[flowchartsman]

Just now seeing this, I'll get into it.