title: Incorrect latest state for a particular key with SinkConnector to ElasticSearch from Kafka Streams
body: <p>So I have an existing MySql database that I want to start tracking with Kafka and presenting data in Elastic.
  Created a topic, created a JDBC connector from a table to that topic (no key) tracking incremental+timestamp, created a SinkConnector to Elastic with MerchantID as a key.
  I have multiple entries for the same MerchantID, Elastic only sees one (as expected) but the current state, from that SinkConnector point of view, is not consistent. Some merchants have the latest id/date, as expected, some others have a random state.
  Imagine the following table existed before I created the connectors to integrate it with Kafka:</p>
  <pre><code>ID  | MerchantID | date          | Balance
  1   | ABC        | 2018-03-04    | 101
  2   | DEF        | 2018-03-04    | 221
  3   | ABC        | 2020-06-22    | 10
  4   | DEF        | 2020-06-23    | 1022
  5   | DEF        | 2020-07-04    | 891
  6   | DEF        | 2021-08-02    | 771
  7   | ABC        | 2021-12-15    | 528
  8   | DEF        | 2021-12-16    | 1221
  </code></pre>
  <p>After all configured, I have <p>
  GET /DEMO-balances-new/_doc/ABC  ---&gt; &quot;ID&quot; : 7 (expected) <p>
  GET /DEMO-balances-new/_doc/DEF  ---&gt; &quot;ID&quot; : 5 (shouldn't it be 8?)</p>
  <p>Code used:</p>
  <pre><code>name=demo-connector
  connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
  connection.url=jdbc:mysql://db-service/test?useSSL=false
  connection.user=
  connection.password=
  tasks.max=1
  topic.prefix=DEMO-
  table.whitelist=balances
  key.converter.schemas.enable=true
  key.converter=io.confluent.connect.avro.AvroConverter
  key.converter.schema.registry.url=http://kafkaschemaregistry-service:8081
  value.converter.schemas.enable=true
  value.converter=io.confluent.connect.avro.AvroConverter
  value.converter.schema.registry.url=http://kafkaschemaregistry-service:8081
  mode=timestamp+incrementing
  timestamp.column.name=date
  incrementing.column.name=id
  
  name=demo-demo-sinkconnector
  connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
  connection.url=https://elasticsearch:9200
  tasks.max=1
  key.ignore=false
  schema.ignore=false
  type.name=_doc
  topics=DEMO-balances-new
  key.converter=io.confluent.connect.avro.AvroConverter
  key.converter.schema.registry.url=http://kafkaschemaregistry-service:8081
  value.converter=io.confluent.connect.avro.AvroConverter
  value.converter.schema.registry.url=http://kafkaschemaregistry-service:8081
  transforms=createKey,extractKey
  transforms.extractKey.type=org.apache.kafka.connect.transforms.ExtractField$Key
  transforms.extractKey.field=MerchantID
  transforms.createKey.type=org.apache.kafka.connect.transforms.ValueToKey
  transforms.createKey.fields=MerchantID
  </code></pre>
  <p>Note: DEMO-balances-new is a stream with a select of few columns from the main topic, VALUE_FORMAT='AVRO', no key defined</p>
  <p>What am I doing wrong here?</p>
  <p>Thanks</p>