Hello I m trying a basic Kafka source sink example It insert Apache Flink #troubleshooting

Hello, I'm trying a basic Kafka source/sink exampl...

Jaehyeon Kim

07/20/2023, 9:12 PM

Hello, I'm trying a basic Kafka source/sink example. It inserts records from a source Kafka topic into a sink topic and both the tables are partitioned by a column called ticker. The table creation definitions can be found in the attached file. When I run the script, I see the following error. Could you please inform how to fix this issue?

Copy code

pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Table 'default_catalog.default_database.sink_table' is a partitioned table, but the underlying DynamicTableSink doesn't implement the SupportsPartitioning interface.
        at org.apache.flink.table.planner.connectors.DynamicSinkUtils.validatePartitioning(DynamicSinkUtils.java:481)
        at org.apache.flink.table.planner.connectors.DynamicSinkUtils.prepareDynamicSink(DynamicSinkUtils.java:398)
        at org.apache.flink.table.planner.connectors.DynamicSinkUtils.convertSinkToRel(DynamicSinkUtils.java:197)
        at org.apache.flink.table.planner.connectors.DynamicSinkUtils.convertSinkToRel(DynamicSinkUtils.java:170)
        at org.apache.flink.table.planner.delegation.PlannerBase.$anonfun$translateToRel$1(PlannerBase.scala:257)

ex.py

sap1ens

07/20/2023, 9:18 PM

Just like the error message says, Kafka sink doesn’t support partitioning clause (

PARTITIONED BY (ticker)

). See this instead. Data can also be pre-partitioned in the DataStream API.

Jaehyeon Kim

07/20/2023, 9:33 PM

Thank you @sap1ens! I took it from a kinesis connector example and didn't know if it is unsupported by the kafka connector. For this example, it is sufficient if all records in a source partition are inserted into a sink partition. When I set

sink.partitioner

value to fixed, it seems to work as expected while the default option doesn't seem to guarantee it. Could you please double check if my understanding is correct?

Martijn Visser

07/20/2023, 9:56 PM

What’s the reason that you want to partition?

Martijn Visser

07/20/2023, 9:58 PM

You can use

sink.partitioner

for Kafka https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/kafka/#sink-partitioning

Jaehyeon Kim

07/20/2023, 10:00 PM

Not a particular reason. The original example based on the kinesis connector created a partitioned table and I followed using the kafka connector. Using

sink.partitioner

value to fixed, records seems to be kept in the same partition although the number gets different, which is okay. Just wanted to double check.

👍 1

Jaehyeon Kim

07/23/2023, 10:06 PM

It works with a local cluster with docker but not in MSK. When setting to fixed, all records are inserted into a single partition.

Martijn Visser

07/24/2023, 6:41 AM

I can't help with that issue unfortunately, I don't know MSK

Open in Slack

Previous Next