https://flink.apache.org/ logo
Join Slack
Powered by
in flink 1.19.0, is there any issue with simple pi...
# troubleshooting
g
in flink 1.19.0, is there any issue with simple pipeline kafka source->sql transformation-> kafka sink in k8s kubernetes flink operator, atleast once guarantee is failing and at time of kubernetes consolidation, data is getting dropped in flink, the pipeline is using processing time 1to 1 sink only, please help
r
You mentioned that this is performing a SQL transformation -- is this a Flink SQL job or using the Datastream API? It shouldn't make much of a difference. I've run multiple jobs using the Flink Operator that all had XO processing guarantees, so I'm not aware of a specific issue with anything internal that's a problem. I'd lean towards configuration and ensuring that all of your related Kafka/Flink bits have the appropriate configuration to ensure XO (e.g., transactional ids, etc.). You also mentioned that data is getting dropped -- have you confirmed this in Kafka? I've seen issues where data is being written but not confirmed (via transactions). You can verify this using the built-in Kafka consumer CLI by adjusting the isolation level (either 
read_uncommitted
or 
read_committed
).
g
i am using flink sql, giving here entire job, we are not using kafka transaction
consolidation is happening daily night, task manager is getting shutting down daily night, and restart, but drop is happening un frequently
consolidation is happening only in eks(flink is deployed there), kafka is untouched as it is msk (so nothing is happening there)
sharing sanitized plan here, removed domain specific knowledge
Copy code
{
  "nodes": [
    {
      "id": 2,
      "type": "Source: generic_kafka_source[1]",
      "pact": "Data Source",
      "contents": "[1]:TableSourceScan(table=[[default_catalog, default_database, generic_kafka_source, metadata=[partition, headers, offset, timestamp]]], fields=[before, after, source, transaction, op, ts_ms, ts_us, ts_ns, __partition, headers, __offset, __ts])",
      "parallelism": 9
    },
    {
      "id": 3,
      "type": "MiniBatchAssigner[2]",
      "pact": "Operator",
      "contents": "[2]:MiniBatchAssigner(interval=[5000ms], mode=[ProcTime])",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 2,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 4,
      "type": "Calc[3]",
      "pact": "Operator",
      "contents": "[3]:Calc(select=[before, source, transaction, ts_ms, ts_us, ts_ns, CAST(before.id AS CHAR(36)) AS id], where=[(op = 'd')])",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 3,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 5,
      "type": "LookupJoin[4]",
      "pact": "Operator",
      "contents": "[4]:LookupJoin(table=[default_catalog.default_database.generic_jdbc_lookup_table], joinType=[LeftOuterJoin], lookup=[id=id], select=[before, source, transaction, ts_ms, ts_us, ts_ns, id, id0])",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 4,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 6,
      "type": "Calc[5]",
      "pact": "Operator",
      "contents": "[5]:Calc(select=[CAST(CAST(ROW(before.col_1, before.col_2, ... <35_more_schema_fields> ...) AS RecordType(TYPE col_1, TYPE col_2, ... <35_more_schema_types> ...)) AS RecordType(TYPE col_1, TYPE col_2, ... <35_more_schema_types> ...)) AS before, null:RecordType(TYPE col_1, TYPE col_2, ... <35_more_schema_types> ...) AS after, source, CAST('d' AS VARCHAR(2147483647)) AS op, ts_ms, ts_us, ts_ns, transaction, CAST(id AS VARCHAR(2147483647)) AS id])",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 5,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 7,
      "type": "Calc[6]",
      "pact": "Operator",
      "contents": "[6]:Calc(select=[after, source, transaction, op, ts_ms, ts_us, ts_ns, CASE((op = 'd'), CAST(before.id AS CHAR(36)), CAST(after.id AS CHAR(36))) AS id], where=[(op <> 'd')])",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 3,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 8,
      "type": "LookupJoin[7]",
      "pact": "Operator",
      "contents": "[7]:LookupJoin(table=[default_catalog.default_database.generic_jdbc_lookup_table], joinType=[LeftOuterJoin], lookup=[id=id], select=[after, source, transaction, op, ts_ms, ts_us, ts_ns, id, id0, looked_up_col_1])",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 7,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 9,
      "type": "Calc[8]",
      "pact": "Operator",
      "contents": "[8]:Calc(select=[null:RecordType(TYPE col_1, TYPE col_2, ... <35_more_schema_types> ...) AS before, CAST(CAST(ROW(after.col_1, after.col_2, ... <...>, CASE((after.col_33 = '__debezium_unavailable_value'), CAST(looked_up_col_1 AS VARCHAR(2147483647)), after.col_33), ... <...>, after.col_37) AS RecordType(TYPE col_1, TYPE col_2, ... <35_more_schema_types> ...)) AS RecordType(TYPE col_1, TYPE col_2, ... <35_more_schema_types> ...)) AS after, source, op, ts_ms, ts_us, ts_ns, transaction, CAST(id AS VARCHAR(2147483647)) AS id])",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 8,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 12,
      "type": "*anonymous_kafka$2*[10]: Writer",
      "pact": "Operator",
      "contents": "*anonymous_kafka$2*[10]: Writer",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 6,
          "ship_strategy": "FORWARD",
          "side": "second"
        },
        {
          "id": 9,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    },
    {
      "id": 14,
      "type": "*anonymous_kafka$2*[10]: Committer",
      "pact": "Operator",
      "contents": "*anonymous_kafka$2*[10]: Committer",
      "parallelism": 9,
      "predecessors": [
        {
          "id": 12,
          "ship_strategy": "FORWARD",
          "side": "second"
        }
      ]
    }
  ]
}
let me know if you need any more log, i can provide that as well
@Rion Williams have you checked it?
2 Views
Open in Slack
PreviousNext
Powered by