Why Flink Table SQL API upsert-kafka sink connector doesn't create a log compacted topic?

I'm trying to replicate Flink's upsert-kafka connector example.

Using the following input:

event_id,user_id,page_id,user_region,viewtime
e0,1,11,TR,2022-01-01T13:26:41.298Z
e1,1,22,TR,2022-01-02T13:26:41.298Z
e2,2,11,AU,2022-02-01T13:26:41.298Z

and created a topic, whose event structure looks like the following:

key: {"event_id":"e2"}, 
value: {"event_id": "e2", "user_id": 2, "page_id": 11, "user_region": "AU", "viewtime": "2022-02-01T13:26:41.298Z"}

Using the following kafka upstream, kafka-upsert sink logic:

CREATE TABLE pageviews_per_region (
  user_region STRING,
  pv BIGINT,
  uv BIGINT,
  PRIMARY KEY (user_region) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'pageviews_per_region',
  'properties.bootstrap.servers' = '...',
  'key.format' = 'json',
  'value.format' = 'json'
);

CREATE TABLE pageviews (
  user_id BIGINT,
  page_id BIGINT,
  viewtime TIMESTAMP,
  user_region STRING,
  WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'pageviews',
  'properties.bootstrap.servers' = '...',
  'format' = 'json'
);

-- calculate the pv, uv and insert into the upsert-kafka sink
INSERT INTO pageviews_per_region
SELECT
  user_region,
  COUNT(*),
  COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_region;

I'm expecting to get just one key for {"user_region":"TR"} with updated pv: 2, however the created topic doesn't seem to be log compacted, hence observing two events for the same user_region:

k: {"user_region":"AU"}, v: {"user_region":"AU","pv":1,"uv":1}
k: {"user_region":"TR"}, v: {"user_region":"TR","pv":2,"uv":1}
k: {"user_region":"TR"}, v: {"user_region":"TR","pv":1,"uv":1}

Isn't the upsert-kafka connector supposed to create a log compacted topic for this use case or is it the developer's responsibility to update the topic configuration?

Another option may be I have misinterpreted something or did a mistake. Looking forward to hear your thoughts. Thanks.

When you use CREATE TABLE to create a table for use with Flink SQL, you are describing how to interpret an existing data store as a table. In other words, you are creating metadata in a Flink catalog. It is Kafka that creates the topic on first access, and it's the developer's responsibility to adjust the log configuration to use a compaction strategy.