kill the downstream kafka cause the cdc server panic

[3AceShowHand]

What did you do?

kill -9 kafka server

What did you expect to see?

the cdc server works normally.

What did you see instead?

panic: runtime error: invalid memory address or nil pointer dereference
        panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x2 addr=0x18 pc=0x1075a192c]

goroutine 5116 [running]:
github.com/pingcap/tiflow/cdc/sink/ddlsink/mq.NewKafkaDDLSink.func1()
        github.com/pingcap/tiflow/cdc/sink/ddlsink/mq/kafka_ddl_sink.go:62 +0x2c
panic({0x1085afc80, 0x10be65090})
        runtime/panic.go:890 +0x248
github.com/pingcap/tiflow/pkg/sink/kafka.AdjustOptions({0x108c18630, 0x1400e04ff20}, {0x0?, 0x0?}, 0x14003faf680, {0x140031f2379, 0x16})
        github.com/pingcap/tiflow/pkg/sink/kafka/options.go:459 +0x44
github.com/pingcap/tiflow/cdc/sink/ddlsink/mq.NewKafkaDDLSink({0x108c18630, 0x1400e04ff20}, 0x2?, 0x14018bf9720, 0x108bd1fa0, 0x108bd1e60)
        github.com/pingcap/tiflow/cdc/sink/ddlsink/mq/kafka_ddl_sink.go:66 +0x250
github.com/pingcap/tiflow/cdc/sink/ddlsink/factory.New({0x108c18630, 0x1400e04ff20}, {0x140031f2362?, 0x108bdb210?}, 0x1084e6cc0?)
        github.com/pingcap/tiflow/cdc/sink/ddlsink/factory/factory.go:50 +0x2e4
github.com/pingcap/tiflow/cdc/owner.ddlSinkInitializer({0x108c18588, 0x140047da690}, 0x14001251d40)
        github.com/pingcap/tiflow/cdc/owner/ddl_sink.go:121 +0x1b4
github.com/pingcap/tiflow/cdc/owner.(*ddlSinkImpl).run.func1()
        github.com/pingcap/tiflow/cdc/owner/ddl_sink.go:154 +0x98
created by github.com/pingcap/tiflow/cdc/owner.(*ddlSinkImpl).run
        github.com/pingcap/tiflow/cdc/owner/ddl_sink.go:150 +0xe8

Versions of the cluster

Upstream TiDB cluster version (execute SELECT tidb_version(); in a MySQL client):

(paste TiDB cluster version here)

Upstream TiKV version (execute tikv-server --version):

(paste TiKV version here)

TiCDC version (execute cdc version):

master