// Module included in the following assemblies:
//
// assembly-config.adoc

[id='proc-listing-connector-offsets-{context}']
= Listing connector offsets

[role="_abstract"]
To track connector offsets using `KafkaConnector` resources, add the `listOffsets` configuration. 
The offsets, which keep track of the flow of data, are written to a config map specified in the configuration. 
If the config map does not exist, Strimzi creates it. 

After the configuration is in place, annotate the `KafkaConnector` resource to write the list to the config map.

Sink connectors use Kafka's standard consumer offset mechanism, while source connectors store offsets in a custom format within a Kafka topic.

* For sink connectors, the list shows Kafka topic partitions and the last committed offset for each partition.
* For source connectors, the list shows the source system’s partition and the last offset processed. 

.Prerequisites

* The Cluster Operator is running.

.Procedure

. Edit the `KafkaConnector` resource for the connector to include the `listOffsets` configuration.
+
.Example configuration to list offsets
[source,yaml,subs="+attributes"]
----
apiVersion: {KafkaConnectorApiVersion}
kind: KafkaConnector
metadata:
  name: my-source-connector
  labels:
    strimzi.io/cluster: my-connect-cluster
spec:
  listOffsets:
    toConfigMap: # <1>
      name: my-connector-offsets # <2>
  # ...  
----
<1> The reference to the config map where the list of offsets will be written to. 
<2> The name of the config map, which is named `my-connector-offsets` in this example. 

. Run the command to write the list to the config map by annotating the `KafkaConnector` resource:
+
[source,shell]
----
kubectl annotate kafkaconnector my-source-connector strimzi.io/connector-offsets=list -n <namespace>
----
+
The annotation remains until either the list operation succeeds or it is manually removed from the resource.

. After the `KafkaConnector` resource is updated, use the following command to check if the config map with the offsets was created:
+
[source,shell]
----
kubectl get configmap my-connector-offsets -n <namespace>
----

. Inspect the contents of the config map to verify the offsets are being listed:
+
[source,shell]
----
kubectl describe configmap my-connector-offsets -n <namespace>
----
+
Strimzi puts the offset information into the `offsets.json` property. 
This does not overwrite any other properties when updating an existing config map.
+
--
.Example source connector offset list
[source,yaml,subs="+attributes"]
----
apiVersion: v1
kind: ConfigMap
metadata:
  # ...
  ownerReferences: # <1>
  - apiVersion: {KafkaConnectApiVersion}
    blockOwnerDeletion: false
    controller: false
    kind: KafkaConnector
    name: my-source-connector
    uid: 637e3be7-bd96-43ab-abde-c55b4c4550e0
  resourceVersion: "66951"
  uid: 641d60a9-36eb-4f29-9895-8f2c1eb9638e
data: 
  offsets.json: |-
    {
      "offsets" : [ {
        "partition" : {
          "filename" : "/data/myfile.txt" # <2>
        },
        "offset" : {
          "position" : 15295 # <3>
        }
      } ]
    } 
----
<1> The owner reference pointing to the `KafkaConnector` resource for the source connector. 
To provide a custom owner reference, create the config map in advance and set the owner reference. 
<2> The source partition, represented by the filename `/data/myfile.txt` in this example for a file-based connector.
<3> The last processed offset position in the source partition.
--
+
--
.Example sink connector offset list
[source,yaml,subs="+attributes"]
----
apiVersion: v1
kind: ConfigMap
metadata:
  # ...
  ownerReferences: # <1>
  - apiVersion: {KafkaConnectApiVersion}
    blockOwnerDeletion: false
    controller: false
    kind: KafkaConnector
    name: my-sink-connector
    uid: 84a29d7f-77e6-43ac-bfbb-719f9b9a4b3b
  resourceVersion: "79241"
  uid: 721e30bc-23df-41a2-9b48-fb2b7d9b042c
data: 
  offsets.json: |-
    {
      "offsets": [
        {
          "partition": {
            "kafka_topic": "my-topic", # <2>
            "kafka_partition": 2 # <3>
          },
          "offset": {
            "kafka_offset": 4 # <4>
          }
        }
      ]
    }
----
<1> The owner reference pointing to the `KafkaConnector` resource for the sink connector. 
<2> The Kafka topic that the sink connector is consuming from.
<3> The partition of the Kafka topic.
<4> The last committed Kafka offset for this topic and partition.
--