In order to make your issue reports as actionable as possible, please provide the following information, depending on the issue type.

Bug report

For bug reports, provide this information, please:

What Debezium connector do you use and what version?

2.2.0 Final (but I have validated the 2.3 latest code has the same bug)

What is the connector configuration?

This bug is in PartitionRouting only, so I only provide the relevant configuration

```

"transforms": "PartitionRouting",
"transforms.PartitionRouting.type": "io.debezium.transforms.partitions.PartitionRouting",
"transforms.PartitionRouting.partition.payload.fields": "change.foo",
"transforms.PartitionRouting.partition.topic.num": "%d",
"transforms.PartitionRouting.predicate": "mytopic",
"predicates": "mytopic",
"predicates.mytopic.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
"predicates.mytopic.pattern": "mytopic*"

```

What is the captured database version and mode of depoyment?

(E.g. on-premises, with a specific cloud provider, etc.)

AWS RDS Postgres 14, MSK Connect

What behaviour do you expect?

Partition routing will ignore nonexistent field in the payload as document described in https://debezium.io/documentation/reference/stable/transformations/partition-routing.html

```

If a specified field is not present in the event message, the SMT ignores it.

```

What behaviour do you see?

A record payload that doesn't contain the `foo` field cause unhandled NullPointerException that crashes the whole connect plugin.

Do you see the same behaviour using the latest relesead Debezium version?

(Ideally, also verify with latest Alpha/Beta/CR version)

yep, check this line https://github.com/debezium/debezium/blob/main/debezium-core/src/main/java/io/debezium/transforms/partitions/PartitionRouting.java#L197

it will throw the same NullPointerException from Optional.of(null)

Do you have the connector logs, ideally from start till finish?

(You might be asked later to provide DEBUG/TRACE level log)

Stack trace:

```

2023-06-08 22:06:36,933 ERROR  ||  WorkerSourceTask{id=tdb_wjzyq_testremotequeue_connection-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted   [org.apache.kafka.connect.runtime.WorkerTask]
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:230)
    at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:156)
    at org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:50)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.sendRecords(AbstractWorkerSourceTask.java:391)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.execute(AbstractWorkerSourceTask.java:359)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
    at org.apache.kafka.connect.runtime.AbstractWorkerSourceTask.run(AbstractWorkerSourceTask.java:75)
    at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:177)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: io.debezium.DebeziumException: Unprocessable message Struct{after=Struct

{id=7,type=XXX,value=0}

,source=Struct

{version=2.2.0.Final,connector=postgresql,name=tdb_wjzyq_testremotequeue,ts_ms=1686261996410,db=tdb_wjzyq_testremotequeue,sequence=["175770232","175773104"],schema=tada,table=datatable,txId=55351,lsn=175773104}

,op=c,ts_ms=1686261996633}
     at io.debezium.transforms.partitions.PartitionRouting.apply(PartitionRouting.java:128)
     at org.apache.kafka.connect.runtime.PredicatedTransformation.apply(PredicatedTransformation.java:56)
     at org.apache.kafka.connect.runtime.TransformationChain.lambda$apply$0(TransformationChain.java:50)
     at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:180)
     at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:214)
     ... 13 more
 Caused by: java.lang.NullPointerException
     at java.base/java.util.Objects.requireNonNull(Objects.java:221)
     at java.base/java.util.Optional.<init>(Optional.java:107)
     at java.base/java.util.Optional.of(Optional.java:120)
     at io.debezium.transforms.partitions.PartitionRouting.toValue(PartitionRouting.java:143)
     at io.debezium.transforms.partitions.PartitionRouting.lambda$apply$0(PartitionRouting.java:112)
     at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195)
     at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1655)
     at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
     at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
     at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
     at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
     at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:578)
     at io.debezium.transforms.partitions.PartitionRouting.apply(PartitionRouting.java:115)
     ... 17 more

```

How to reproduce the issue using our tutorial deployment?

You will need to create a table `tada.datatable` in postgres, with schema:

id, type, value, foo (optional)

Then insert a record doesn't provide the value for `foo`

Feature request or enhancement

For feature requests or enhancements, provide this information, please:

Which use case/requirement will be addressed by the proposed feature?

PartitionRouting

Implementation ideas (optional)

<Your answer>