feat: Add consumer offsets to DESCRIBE EXTENDED

[jeqo]

Description

Fix #3604

Testing done

• Unit tests covering new options, with consumer offsets

Reviewer checklist

• Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
• Ensure relevant issues are linked (description should include text like "Fixes #")

Missing functionality

• Get groupId from Source
• Set timeout for Kafka Admin operations

[ghost]

It looks like @jeqo hasn't signed our Contributor License Agreement, yet.

The purpose of a CLA is to ensure that the guardian of a project's outputs has the necessary ownership or grants of rights over all contributions to allow them to distribute under the chosen licence.
Wikipedia

You can read and sign our full Contributor License Agreement here.

Once you've signed reply with [clabot:check] to prove it.

Appreciation of efforts,

clabot

[jeqo]

Current output:

ksql> describe extended orders;

Name                 : ORDERS
Type                 : STREAM
Timestamp field      : Not set - using <ROWTIME>
Key format           : KAFKA
Value format         : JSON
Kafka topic          : test_topic (partitions: 12, replication: 1)
Statement            : CREATE STREAM orders (
    ROWKEY INT KEY,
    ORDERUNITS double
)
WITH (
    kafka_topic='test_topic',
    partitions=12,
    replicas=1,
    value_format='JSON'
);

 Field      | Type                   

 ROWKEY     | INTEGER          (key) 
 ORDERUNITS | DOUBLE                 


Queries that read from this STREAM
-----------------------------------
CSAS_S1_0 (RUNNING) : CREATE STREAM S1 WITH (KAFKA_TOPIC='S1', PARTITIONS=12, REPLICAS=1) AS SELECT   ORDERS.ROWKEY ROWKEY,   ORDERS.ORDERUNITS ORDERUNITS,   (CASE WHEN (ORDERS.ORDERUNITS < 2.0) THEN 'small' WHEN (ORDERS.ORDERUNITS < 4.0) THEN 'medium' ELSE 'large' END) CASE_RESULT FROM ORDERS ORDERS EMIT CHANGES;

For query topology and execution plan please run: EXPLAIN <QueryId>

Local runtime statistics
------------------------


(Statistics of the local KSQL server interaction with the Kafka topic test_topic)

Consumer Group       : _confluent-ksql-default_query_CSAS_S1_0
Kafka topic          : test_topic

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 8          | 8      | 0   
 1         | 0            | 0          | 0      | 0   
 2         | 0            | 0          | 0      | 0   
 3         | 0            | 0          | 0      | 0   
 4         | 0            | 2          | 2      | 0   
 5         | 0            | 0          | 0      | 0   
 6         | 0            | 0          | 0      | 0   
 7         | 0            | 0          | 0      | 0   
 8         | 0            | 0          | 0      | 0   
 9         | 0            | 0          | 0      | 0   
 10        | 0            | 0          | 0      | 0   
 11        | 0            | 0          | 0      | 0   

[agavra]

Thanks @jeqo! This is a really good first stab at solving this, and I think it'll be a really useful tool to have when looking at ksqlDB applications operationally. I didn't have time to give this a full review, but I figured I give you some first round comments in the meantime.

[jeqo]

[clabot:check]

[ghost]

@confluentinc It looks like @jeqo just signed our Contributor License Agreement. 👍

Always at your service,

clabot

[agavra]

mostly LGTM, sorry for the delayed review cycles! a few comments still left

[big-andy-coates]

And then a small enhancement request, if you don't mind, to make this much more useful when there are LARGE numbers of partitions...

Can we add a 'max lag' summary at the top please? That way if you have 100 or 1000 partitions, you don't need to scan down to find the worst lag: you can just look at this value. Of course, you'd still need to scan down to find which partition was lagging, but at least you can quickly see how badly, or not, the group is lagging.

For example:

Consumer Group       : _confluent-ksql-default_query_CSAS_S1_0
Kafka topic          : test_topic
Max lag:             : 8

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 8          | 8      | 0   
 1         | 9            |24          | 16     | 8  
 2         | 0            | 0          | 0      | 0   
 3         | 0            | 0          | 0      | 0   
 4         | 0            | 2          | 2      | 0   
 5         | 0            | 0          | 0      | 0   
 6         | 0            | 0          | 0      | 0   
 7         | 0            | 10         | 8      | 2   
 8         | 0            | 0          | 0      | 0   
 9         | 0            | 0          | 0      | 0   
 10        | 0            | 0          | 0      | 0   
 11        | 0            | 0          | 0      | 0 

[big-andy-coates]

Hi @jeqo

Awesome that you've created this PR - very much appreciated. I've added some comments, nits and suggestions below.

As it stands, I'm not sure the design is quite right. I'm not sure if the relationship between topic and consumer groups are right. If I'm reading the code correctly, (apologies, I've not had time to pull it locally and debug to check), then it's comparing the partition offsets of a source's sink topic with the consumer group offsets of queries that write into the source. I believe this is wrong.

Let's look at an example to work it through:

CREATE STREAM S1 (... some columns... )
   WITH (kafka_topic='s1', ...);

CREATE STREAM S2 (... some columns... )
   WITH (kafka_topic='s2', ...);

CREATE STREAM OUTPUT WITH (
     kafka_topic='op'
   ) AS 
     SELECT *
     FROM S1
       JOIN S2 WITHIN 10 SECONDS ON S1.id = S2.id;

Then we run:

DESCRIBE EXTENDED OUTPUT;

This means:

• our DataSource is OUTPUT.
• it's kafkaTopicName is op.
• and it has a single RunningQuery.

Importantly, the consumer group of the RunningQuery will be consuming from topics s1 and s2, and NOT op.

So we can not compare the offsets of the consumer group with the offsets of op. We must compare the offsets of the consumer group with the offsets of the topic-partitions the consumer group is consuming!

If we add in a second query:

CREATE STREAM S3 (... some columns... )
   WITH (kafka_topic='s3', ...);

INSET INTO OUTPUT
   SELECT * FROM S3;

Now there will be two RunningQuerys for OUTPUT. The second consumer group will be consuming from s3 only.

So we need to change the logic to compare offsets of the right topics. I would suggest the following logic:

1. Iterate over all RunningQuery
   a. calculate queryApplicationId
   b. describe the consumer group and get the map of TopicPartition -> offset.
2. Once you've iterated over all running queries you can:
   a. make a single request to get the earliest offset of ALL topic partitions we're interested in, i.e. all partitions of s1, s2 and s3 from the example above.
   b. make a second single request to get the latest offsets of all the tps.

This way, we minimise calls to the brokers.

[jeqo]

@big-andy-coates It was a bit confusing at the beginning to see how the concepts fit together internally, but your example makes it so much clearer, thanks for this!

I'd expect to go through this around next week.

[jeqo]

Output starts to look better:

Consumer Group       : _confluent-ksql-default_query_CSAS_OUTPUT_0
Kafka topic          : S1
Max lag              : 0

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 0          | 0      | 0   
 1         | 0            | 0          | 0      | 0   
 2         | 0            | 0          | 0      | 0   
 3         | 0            | 1          | 1      | 0   
 4         | 0            | 0          | 0      | 0   
 5         | 0            | 0          | 0      | 0   
 6         | 0            | 0          | 0      | 0   
 7         | 0            | 0          | 0      | 0   
 8         | 0            | 0          | 0      | 0   
 9         | 0            | 0          | 0      | 0   
 10        | 0            | 0          | 0      | 0   
 11        | 0            | 0          | 0      | 0   


Consumer Group       : _confluent-ksql-default_query_CSAS_OUTPUT_0
Kafka topic          : S2
Max lag              : 0

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 0          | 0      | 0   
 1         | 0            | 0          | 0      | 0   
 2         | 0            | 0          | 0      | 0   
 3         | 0            | 1          | 1      | 0   
 4         | 0            | 0          | 0      | 0   
 5         | 0            | 0          | 0      | 0   
 6         | 0            | 0          | 0      | 0   
 7         | 0            | 0          | 0      | 0   
 8         | 0            | 0          | 0      | 0   
 9         | 0            | 0          | 0      | 0   
 10        | 0            | 0          | 0      | 0   
 11        | 0            | 0          | 0      | 0   


Consumer Group       : _confluent-ksql-default_query_INSERTQUERY_7
Kafka topic          : S3
Max lag              : 0

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 1          | 1      | 0 

Will go through the specifics later.

[jeqo]

@big-andy-coates I managed to cover the new requirements, take a look to the output:

ksql> describe extended output;

/////......

Queries that write from this STREAM
-----------------------------------
CSAS_OUTPUT_0 (RUNNING) : CREATE STREAM OUTPUT WITH (KAFKA_TOPIC='OP', PARTITIONS=12, REPLICAS=1) AS SELECT   S1.NAME NAME,   S1.AGE AGE,   S2.COUNTRY COUNTRY FROM S1 S1 INNER JOIN S2 S2 WITHIN 10 SECONDS ON ((S1.NAME = S2.NAME)) EMIT CHANGES;
INSERTQUERY_7 (RUNNING) : INSERT INTO OUTPUT SELECT * FROM S3;

For query topology and execution plan please run: EXPLAIN <QueryId>

Local runtime statistics
------------------------


(Statistics of the local KSQL server interaction with the Kafka topic OP)

Consumer Groups summary:

Consumer Group       : _confluent-ksql-default_query_CSAS_OUTPUT_0

Kafka topic          : S1
Max lag              : 0

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 0          | 0      | 0   
 1         | 0            | 0          | 0      | 0   
 2         | 0            | 0          | 0      | 0   
 3         | 0            | 3          | 3      | 0   
 4         | 0            | 0          | 0      | 0   
 5         | 0            | 0          | 0      | 0   
 6         | 0            | 0          | 0      | 0   
 7         | 0            | 0          | 0      | 0   
 8         | 0            | 0          | 0      | 0   
 9         | 0            | 0          | 0      | 0   
 10        | 0            | 0          | 0      | 0   
 11        | 0            | 0          | 0      | 0   


Kafka topic          : S2
Max lag              : 0

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 0          | 0      | 0   
 1         | 0            | 0          | 0      | 0   
 2         | 0            | 0          | 0      | 0   
 3         | 0            | 3          | 3      | 0   
 4         | 0            | 0          | 0      | 0   
 5         | 0            | 0          | 0      | 0   
 6         | 0            | 0          | 0      | 0   
 7         | 0            | 0          | 0      | 0   
 8         | 0            | 0          | 0      | 0   
 9         | 0            | 0          | 0      | 0   
 10        | 0            | 0          | 0      | 0   
 11        | 0            | 0          | 0      | 0   


Consumer Group       : _confluent-ksql-default_query_INSERTQUERY_7

Kafka topic          : S3
Max lag              : 0

 Partition | Start Offset | End Offset | Offset | Lag 

 0         | 0            | 3          | 3      | 0   

and when topics are empty for a consumer group, the message you recommend is printed.

[big-andy-coates]

Great work @jeqo - thanks for the contribution!