Pulsar Source Connector to Postgresql

[amin224]

Hello,

I have a running docker container of Pulsar (image: apachepulsar/pulsar-all:2.9.0) and I am trying to connect it to my local machine Postgresql database, but i am getting exceptions.

bin/pulsar-admin source create --destination-topic-name my-topic --source-config-file ./connectors/pulsar-postgres-jdbc-source.yaml

I added a yaml file to the /pulsar/connectors directory:

tenant: "public"
namespace: "default"
name: "debezium-postgres-source"
inputs: [ "my-topic" ]
archive: "connectors/pulsar-io-debezium-postgres-2.9.0.nar"
parallelism: 1

configs:
  database.hostname: "{MY MACHINE IP}"
  database.port: "5432"
  database.user: "postgres"
  database.password: "{mypassword}"
  database.dbname: "postgres"
  database.server.name: "postgres"
  plugin.name: "pgoutput"
  schema.whitelist: "public"
  table.whitelist: "public.pulsar_postgres_jdbc_sink"
  database.history.pulsar.service.url: "pulsar://127.0.0.1:6650"

I made sure I can ping to my local machine from docker container, in addition, telnet to port 5432 and there was successful connection.
Please let me know if I am doing something wrong here are the exceptions:

2023-08-11T10:25:14,611+0000 [function-timer-thread-86-1] ERROR org.apache.pulsar.functions.runtime.process.ProcessRuntime - Health check failed for debezium-postgres-source-0
java.util.concurrent.ExecutionException: io.grpc.StatusRuntimeException: UNAVAILABLE: io exception
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395) ~[?:?]
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999) ~[?:?]
	at org.apache.pulsar.functions.runtime.process.ProcessRuntime.lambda$start$1(ProcessRuntime.java:186) ~[org.apache.pulsar-pulsar-functions-runtime-2.9.0.jar:2.9.0]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.68.Final.jar:4.1.68.Final]
	at java.lang.Thread.run(Thread.java:829) [?:?]
Caused by: io.grpc.StatusRuntimeException: UNAVAILABLE: io exception
	at io.grpc.Status.asRuntimeException(Status.java:533) ~[io.grpc-grpc-api-1.33.0.jar:1.33.0]
	at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:533) ~[io.grpc-grpc-stub-1.33.0.jar:1.33.0]
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:616) ~[io.grpc-grpc-core-1.33.0.jar:1.33.0]
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:69) ~[io.grpc-grpc-core-1.33.0.jar:1.33.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:802) ~[io.grpc-grpc-core-1.33.0.jar:1.33.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:781) ~[io.grpc-grpc-core-1.33.0.jar:1.33.0]
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[io.grpc-grpc-core-1.33.0.jar:1.33.0]
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123) ~[io.grpc-grpc-core-1.33.0.jar:1.33.0]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
	... 1 more
Caused by: io.grpc.netty.shaded.io.netty.channel.AbstractChannel$AnnotatedConnectException: finishConnect(..) failed: Connection refused: /127.0.0.1:42355
Caused by: java.net.ConnectException: finishConnect(..) failed: Connection refused
	at io.grpc.netty.shaded.io.netty.channel.unix.Errors.throwConnectException(Errors.java:124) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.channel.unix.Socket.finishConnect(Socket.java:243) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.doFinishConnect(AbstractEpollChannel.java:672) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.finishConnect(AbstractEpollChannel.java:649) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.epollOutReady(AbstractEpollChannel.java:529) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:465) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:378) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	at io.grpc.netty.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[io.grpc-grpc-netty-shaded-1.33.0.jar:1.33.0]
	... 1 more
2023-08-11T10:25:14,615+0000 [function-timer-thread-86-1] ERROR org.apache.pulsar.functions.runtime.process.ProcessRuntime - Extracted Process death exception
java.lang.RuntimeException: 
	at org.apache.pulsar.functions.runtime.process.ProcessRuntime.tryExtractingDeathException(ProcessRuntime.java:402) ~[org.apache.pulsar-pulsar-functions-runtime-2.9.0.jar:2.9.0]
	at org.apache.pulsar.functions.runtime.process.ProcessRuntime.isAlive(ProcessRuntime.java:389) ~[org.apache.pulsar-pulsar-functions-runtime-2.9.0.jar:2.9.0]
	at org.apache.pulsar.functions.runtime.RuntimeSpawner.lambda$start$0(RuntimeSpawner.java:86) ~[org.apache.pulsar-pulsar-functions-runtime-2.9.0.jar:2.9.0]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.68.Final.jar:4.1.68.Final]
	at java.lang.Thread.run(Thread.java:829) [?:?]
2023-08-11T10:25:14,616+0000 [function-timer-thread-86-1] ERROR org.apache.pulsar.functions.runtime.RuntimeSpawner - public/default/debezium-postgres-source Function Container is dead with following exception. Restarting.
java.lang.RuntimeException: 
	at org.apache.pulsar.functions.runtime.process.ProcessRuntime.tryExtractingDeathException(ProcessRuntime.java:402) ~[org.apache.pulsar-pulsar-functions-runtime-2.9.0.jar:2.9.0]
	at org.apache.pulsar.functions.runtime.process.ProcessRuntime.isAlive(ProcessRuntime.java:389) ~[org.apache.pulsar-pulsar-functions-runtime-2.9.0.jar:2.9.0]
	at org.apache.pulsar.functions.runtime.RuntimeSpawner.lambda$start$0(RuntimeSpawner.java:86) ~[org.apache.pulsar-pulsar-functions-runtime-2.9.0.jar:2.9.0]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [io.netty-netty-common-4.1.68.Final.jar:4.1.68.Final]
	at java.lang.Thread.run(Thread.java:829) [?:?]

bin/pulsar-admin sources status \
>   --tenant public \
>   --namespace default \
>   --name debezium-postgres-source
{
  "numInstances" : 1,
  "numRunning" : 0,
  "instances" : [ {
    "instanceId" : 0,
    "status" : {
      "running" : false,
      "error" : "UNAVAILABLE: io exception",
      "numRestarts" : 0,
      "numReceivedFromSource" : 0,
      "numSystemExceptions" : 0,
      "latestSystemExceptions" : [ ],
      "numSourceExceptions" : 0,
      "latestSourceExceptions" : [ ],
      "numWritten" : 0,
      "lastReceivedTime" : 0,
      "workerId" : "c-standalone-fw-localhost-8080"
    }
  } ]
}