PGMoney Type Get Value is Error：

@Override
public PGmoney asMoney() {
try {
final String value = asString();
if (value != null && value.startsWith("-"))

{ final String negativeMoney = "(" + value.substring(1) + ")"; return new PGmoney(negativeMoney); }

return new PGmoney(asString());
} catch (final Exception e) {
LOGGER.error("Failed to parse money {}, {}", asString(), e);
throw new ConnectException(e);
}
}

stack log:

org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped. at io.debezium.connector.base.ChangeEventQueue.throwProducerFailureIfPresent(ChangeEventQueue.java:171) at io.debezium.connector.base.ChangeEventQueue.poll(ChangeEventQueue.java:151) at io.debezium.connector.postgresql.PostgresConnectorTask.poll(PostgresConnectorTask.java:204) at org.apache.kafka.connect.runtime.WorkerSourceTask.poll(WorkerSourceTask.java:265) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.kafka.connect.errors.ConnectException: java.lang.NullPointerException at io.debezium.connector.postgresql.connection.AbstractColumnValue.asMoney(AbstractColumnValue.java:142) at io.debezium.connector.postgresql.connection.pgproto.PgProtoColumnValue.asMoney(PgProtoColumnValue.java:223) at io.debezium.connector.postgresql.connection.ReplicationMessageColumnValueResolver.resolveValue(ReplicationMessageColumnValueResolver.java:140) at io.debezium.connector.postgresql.connection.pgproto.PgProtoReplicationMessage.getValue(PgProtoReplicationMessage.java:132) at io.debezium.connector.postgresql.connection.pgproto.PgProtoReplicationMessage$1.getValue(PgProtoReplicationMessage.java:112) at io.debezium.connector.postgresql.PostgresChangeRecordEmitter.columnValues(PostgresChangeRecordEmitter.java:170) at io.debezium.connector.postgresql.PostgresChangeRecordEmitter.getOldColumnValues(PostgresChangeRecordEmitter.java:100) at io.debezium.relational.RelationalChangeRecordEmitter.emitDeleteRecord(RelationalChangeRecordEmitter.java:120) at io.debezium.relational.RelationalChangeRecordEmitter.emitChangeRecords(RelationalChangeRecordEmitter.java:52) at io.debezium.connector.postgresql.PostgresChangeRecordEmitter.emitChangeRecords(PostgresChangeRecordEmitter.java:88) at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:151) at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.lambda$execute$0(PostgresStreamingChangeEventSource.java:155) at io.debezium.connector.postgresql.connection.pgproto.PgProtoMessageDecoder.processNotEmptyMessage(PgProtoMessageDecoder.java:67) at io.debezium.connector.postgresql.connection.AbstractMessageDecoder.processMessage(AbstractMessageDecoder.java:33) at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.deserializeMessages(PostgresReplicationConnection.java:451) at io.debezium.connector.postgresql.connection.PostgresReplicationConnection$1.readPending(PostgresReplicationConnection.java:445) at io.debezium.connector.postgresql.PostgresStreamingChangeEventSource.execute(PostgresStreamingChangeEventSource.java:120) at io.debezium.pipeline.ChangeEventSourceCoordinator.lambda$start$0(ChangeEventSourceCoordinator.java:101) ... 5 more Caused by: java.lang.NullPointerException at org.postgresql.util.PGmoney.setValue(PGmoney.java:45) at org.postgresql.util.PGmoney.<init>(PGmoney.java:30) at io.debezium.connector.postgresql.connection.AbstractColumnValue.asMoney(AbstractColumnValue.java:134) ... 22 more

 

Finally:

The task is Failed。Task crash。