- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
更新后出错
[2019-07-29 12:52:23,301] INFO Initializing writer using SQL dialect: PostgreSqlDatabaseDialect (io.confluent.connect.jdbc.sink.JdbcSinkTask:57)
[2019-07-29 12:52:23,303] INFO WorkerSinkTask{id=sink-postgres-0} Sink task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSinkTask:301)
[2019-07-29 12:52:23,367] WARN [Consumer clientId=consumer-1, groupId=connect-sink-postgres] Error while fetching metadata with correlation id 2 : {kafkadad=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient:1023)
[2019-07-29 12:52:23,368] INFO Cluster ID: _gRuX5-0SUu72wzy6PV0Ag (org.apache.kafka.clients.Metadata:365)
[2019-07-29 12:52:23,369] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] Discovered group coordinator INTRIVMPIOT01.xpetize.local:9092 (id: 2147483647 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:675)
[2019-07-29 12:52:23,372] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] Revoking previously assigned partitions [] (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:459)
[2019-07-29 12:52:23,373] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:491)
[2019-07-29 12:52:23,383] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] (Re-)joining group (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:491)
[2019-07-29 12:52:23,482] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] Successfully joined group with generation 1 (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:455)
[2019-07-29 12:52:23,486] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] Setting newly assigned partitions: kafkadad-0 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:290)
[2019-07-29 12:52:23,501] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] Resetting offset for partition kafkadad-0 to offset 0. (org.apache.kafka.clients.consumer.internals.Fetcher:584)
[2019-07-29 12:52:35,338] ERROR WorkerSinkTask{id=sink-postgres-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:177)
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:178)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:104)
at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:487)
at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:464)
at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:320)
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:224)
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:192)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
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: java.lang.NullPointerException
at org.apache.kafka.connect.json.JsonConverter.convertToConnect(JsonConverter.java:701)
at org.apache.kafka.connect.json.JsonConverter.access$000(JsonConverter.java:61)
at org.apache.kafka.connect.json.JsonConverter$12.convert(JsonConverter.java:181)
at org.apache.kafka.connect.json.JsonConverter.convertToConnect(JsonConverter.java:745)
at org.apache.kafka.connect.json.JsonConverter.toConnectData(JsonConverter.java:363)
at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$1(WorkerSinkTask.java:487)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:128)
at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:162)
... 13 more
[2019-07-29 12:52:35,347] ERROR WorkerSinkTask{id=sink-postgres-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:178)
[2019-07-29 12:52:35,347] INFO Stopping task (io.confluent.connect.jdbc.sink.JdbcSinkTask:105)
[2019-07-29 12:52:35,349] INFO [Consumer clientId=consumer-1, groupId=connect-sink-postgres] Member consumer-1-bdbc7035-7625-4701-9ca7-c1ffa6863456 sending LeaveGroup request to coordinator INTRIVMPIOT01.xpetize.local:9092 (id: 2147483647 rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:822)
connect-standalone.properties 文件
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=true
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
plugin.path=/home/kafka/confluent-5.2.1/share/java
connect-post.properties 文件
name=sink-postgres
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=2
topics=kafkada
connection.url=jdbc:postgresql://localhost:5432/kafkadb?
user=postgres&password=postgres
insert.mode=upsert
table.name.format=kafkatable
pk.mode=none
pk.fields=none
auto.create=true
auto.evolve=false
offset.storage.file.filename=/tmp/post-sink.offsets
上面的错误是我通过apache kafka./bin/connect-standalone.sh config/connect-standalone.properties config.postgresql.properties导致的。
然后,我尝试并实现了这个链接中提到的流程:
但是,这里的数据是使用 avro 从 Python 代码生成的。但就我而言,我已经有了来自 kafka 主题中的传感器(JSON 格式)的数据,我想将这些数据发送到 postgreSQL,而不是通过代码生成数据。
那么,我怎样才能实现从 kafka 主题向 postgreSQL 发送数据的流程。
我已经分享了我的属性文件,如果需要更正,请告诉我。我正在发送简单的 json 数据,如 "{"cust_id": 1313131, "month": 12, "expenses": 1313.13}"我也尝试发送这种类型的数据,但仍然存在错误
示例 json 数据
{
"schema": {
"type": "struct",
"fields": [
{
"type": "int32",
"optional": false,
"field": "customer_id"
},
{
"type": "int32",
"optional": true,
"field": "month"
},
{
"type": "string",
"optional": true,
"field": "amount_paid"
}
],
"optional": false,
"name": "msgschema"
},
"payload": {
"cust_id": 13,
"month": 12,
"expenses": 1313.13
}
}
我有一个名为 kafkatable 的表,它的列名称为 (customer_id, month, amount_paid) 使用
创建“创建表 kafkatable( customer_id int8, month int4, amount_paid decimal(9,2) );”
最佳答案
我通过进行以下更改解决了这个错误
关于postgresql - 以 JSON 格式将数据从 Kafka 主题推送到 PostgreSQL,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/57158092/
这个问题已经有答案了: Cannot create AlertDialog: AppCompat error (2 个回答) 已关闭 6 年前。 当我在列表项中调用警报对话框时,我的应用程序崩溃了。我
我在 Angular 应用程序中安装了 Material UI,现在我收到错误,没有导出的成员 Observable 错误, 我删除了节点模块并重新安装,问题仍然存在 ERROR in node_mo
我有一个架构,其中有两个独立的应用程序。原始来源是一个sql数据库。 App1 监听 CDC 表以跟踪对该数据库中表的更改、规范化和序列化这些更改。它获取这些序列化消息并将它们发送到 Kafka 主题
这个问题在这里已经有了答案: Material Design, AppCompat, and Backwards Compatibility (1 个回答) 关闭 6 年前。 我收到如下错误信息:
我喜欢新 Office 套件和 Visual Studio 上的窗口镶边: 当然,我仍在为 Windows 7 开发应用程序,但我想知道是否有一种快速且简单的方法(阅读:WPF 样式或 Windows
我正在使用 HoloEverywhere-1.6.8。 我有一个基于 Holo.Theme 的自定义主题。 ... 我遇到了下面的崩溃,但它只出现在以下设备上: Galaxy Tab 10.1 P
我正在尝试为 Angular 的 DevExtreme 小部件加载主题。我采用了不同的方法: 在 angular.json 中设置样式但不会产生任何影响: "projects": { "my-proj
我想定义一个 android 样式,它扩展了一个在不同的应用程序/包中定义的样式,而不是作为库导入。 从对android资源的xml引用的定义here : @[:]/ 似乎可以在定义资源的地方指定一个
我正在尝试测试一种制作主题的方法,但我使用的方法并没有给我预期的结果。这是我的设置: drawable/dummy.xml 值/mythemes.xml @style
通过 telnet 使用 IMAP,我希望能够从特定的给定电子邮件中提取主题。现在我知道 fetch 命令负责从电子邮件中获取数据。 我的问题是,如何在不使用对 BODY[HEADER.FIELDS
我刚刚开始使用 RStudio 中的一些新的 knitr 功能。 我已经尝试在 R Markdown 设置对话框中选择几个不同的主题,但这些似乎对我生成的文档的样式没有任何明显的影响。应该,还是我错过
在我的应用程序中,我有多种主题样式(您可以将它们视为不同的、单独的 CSS 样式文件)。我想开始使用 CSS 模块,但我什至不知道如何 import我的第一个文件。 让我们假设以下(简单)目录结构:
有没有一种方法可以在一个 Azure 主题订阅上拥有多个监听客户端,并且它们都接收所有消息?我的理解是订阅的唯一实现是发布的消息仅传递到该订阅上的一个客户端,因为它就像一个队列。 可以使用同一订阅将这
我有一台 super 光滑的显示器,所以白天我可以比深色主题上的代码更好地看到自己的倒影。因此,我认为如果我可以在 vimrc 中有一个简单的 if 开关来根据一天中的时间设置深色主题或浅色主题,那就
我希望在我的 Symfony2 项目中提供基本的主题支持,因此我希望为每个主题提供单独的静态文件(css、js、img)。 我尝试添加 assetic: read_from: %kernel
有没有一种方法可以在一个 Azure 主题订阅上拥有多个监听客户端,并且它们都接收所有消息?我的理解是订阅的唯一实现是发布的消息仅传递到该订阅上的一个客户端,因为它就像一个队列。 可以使用同一订阅将这
在 NES 上有多个处于 WAITING 状态的“Discovery Worker”和“Keep Alive”线程是预期的行为吗? "DiscoveryWorker-10" Id=62 WAITING
我正在尝试找到最适合加载图像的颜色并将其应用到背景中。适应图像并使 UI 感觉更自然。 到目前为止我已经找到了 2 个方案: 1> 平均像素(下面的代码): final Color acclimati
我知道每个请求都由一个 servlet 线程提供服务,但是对于一个用户 session ,两个请求是否可以由两个不同的线程提供服务? 如果上述情况真的发生,那么第一个请求服务线程存储的线程局部变量被第
我无法理解操作栏外观与主题化之间的交互模式。我的应用设置为使用默认主题,我认为它是深色的: 通过应用范围内的样式从应用中删除操作栏会导致主要 Activity 的黑色背景: 没有 and
我是一名优秀的程序员,十分优秀!