This message was deleted.

this is on postgres side, can you share your pg table schema?

Agreed. The error indicates that Postgres is throwing an exception. However, the UUID field doesn’t appear to be getting properly parsed inside the connector.

@Yu Wei Sung my postgres side is like below. so,

uuid

is actually a primary key. since it's part of the requirment.

CREATE TABLE public.jdbc_sink (
	uuid uuid NOT NULL,
	"timestamp" int8 NULL,
	CONSTRAINT jdbc_sink_pkey PRIMARY KEY (uuid)
);

you can try turn on debug mode and trace this line. https://github.com/apache/pulsar/blob/a9d5d25e52710503cc2642ba7a8aadda0a32faae/pul[…]main/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSink.java my suspicion is the data type. in your avro schem, uuid: string but in pg table, it is uuid: uuid. easy troubleshooting step: you can try another table with uuid: text as primary key.

Thanks @Yu Wei Sung, it was thoughtful one that the error was on schema. however i already try it, but still getting the same error. The schema using

pulsar-admin schema get

function.

root@d5da1b9bf4c2:/pulsar# bin/pulsar-admin schemas get pulsar-postgres-jdbc-sink-topic      
{
  "version": 2,
  "schemaInfo": {
    "name": "jdbc_source_output",
    "schema": {
      "type": "record",
      "name": "jdbc_sink",
      "fields": [
        {
          "name": "uuid",
          "type": {
            "type": "string",
            "logicalType": "uuid"
          }
        },
        {
          "name": "timestamp",
          "type": "long"
        }
      ]
    },
    "type": "AVRO",
    "timestamp": 1697675942684,
    "properties": {}
  }
}

report this to gh issue if this doesn’t work.

@Yu Wei Sung @David K i use some altrenative by create some

pulsar function

with python. in summary this pulsar function use

psycopg2

to connect directly to the db and parse some json from the topics, and use

INSERT INTO {table_name} ({columns}) VALUES ({placeholders})

CRUD to insert.

import pulsar
from pulsar import Function
import psycopg2
import json

class PostgresSync(Function):

    def process(self, input, context):
        error_messages = []

        context.get_logger().info(f"Received input: {input}")
        
        conn = None
        cursor = None
        
        try:
            context.get_logger().info("Trying to establish PostgreSQL connection...")
            
            conn = psycopg2.connect(
                host="localhost",
                port=5432,
                dbname="postgres",
                user="postgres",
                password="password"
            )
            cursor = conn.cursor()
            
            context.get_logger().info("Successfully established PostgreSQL connection.")
        except Exception as e:
            err_msg = f"Error while establishing connection to PostgreSQL: {e}"
            context.get_logger().error(err_msg)
            error_messages.append(err_msg)

        try:
            # Parse the content
            data = json.loads(input)
            after_data = data.get("after", {})
            
            # Extract table name and append "_destination"
            table_name = data.get("source", {}).get("table", "") + "_destination"

            # Constructing dynamic query
            columns = ", ".join(after_data.keys())
            placeholders = ", ".join(["%s"] * len(after_data))
            values = tuple(after_data.values())
            
            query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders});"
            
            cursor.execute(query, values)
            conn.commit()
            
            context.get_logger().info("Successfully inserted data into PostgreSQL.")
        except Exception as e:
            err_msg = f"Error during data insertion into PostgreSQL. Data: {input}. Error: {e}"
            context.get_logger().error(err_msg)
            error_messages.append(err_msg)

        try:
            # Close the connection
            cursor.close()
            conn.close()
            context.get_logger().info("Closed PostgreSQL connection successfully.")
        except Exception as e:
            err_msg = f"Error while closing the PostgreSQL connection: {e}"
            context.get_logger().error(err_msg)
            error_messages.append(err_msg)

        return "; ".join(error_messages) if error_messages else input

This approach is not scalable. For starters recreating the database connection every record is inefficient. The connection should be created once and then cached. Secondly the record should be published in a batch rather than individually. However even these changes will not work for 100k+ rows per second.

Architecture wise, it is jdbc, aka connection. You should not use multiple jdbc connections to sink to a single table. Both pulsar and postgres have their own transaction mechanisms. There are many scalability concerns coordinating those rollback commits acknowledge….to consider

sure, thanks for the input. already discard this option. right now i try to understand better the pulsar schema. both AVRO and JSON.