Блог

Я соединяю две таблицы и извлекаю данные в Kafka. Если я использую mode как массовый, то он работает нормально, но если я использую mode как временную метку и даю timestamp.column.name соответственно, тогда это показывает

Моя конфигурация выглядит следующим образом:

     "name": "join-test-31",
    "config": {
        "connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.password": "******",
        "validate.non.null": "true",
        "catalog.pattern": "public",
        "connection.url": "jdbc:postgresql://localhost/test",
        "mode": "timestamp",
        "query": "SELECT "p"."Id" as "p_Id","p"."CreatedDate" as 
         "p_CreatedDate","p"."ClassId" as 
         "p_ClassId","c"."Id" as 
         "c_Id","c"."CreatedDate" as 
         "c_CreatedDate","c"."ClassId" as "c_ClassId" FROM 
         "public"."Policy" "p" left join "public"."ClassDe" 
         "c" on "p"."DocumentNumber" = "c"."DocumentNo"",
        "timestamp.column.name": "p_CreatedDate",
        "key.converter.schemas.enable": "true",
        "topic.prefix": "test_join",
        "connection.user": "postgres",
        "value.converter.schemas.enable": "true",
        "errors.tolerance": "all",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "errors.log.enable": "true",
        "key.converter":"org.apache.kafka.connect.storage.StringConverter"
    }
}