I am facing a weird issue and not able to solve it or get any clue, any pointers would be really helpful… This is the transformation I am trying to do..

DataStream[Event] -> Window -> ProcessWindowFunction -> Datastream[Event2] -> Kafka Sink

Everything works fine but i am seeing in kafka some value is getting shuffled, i.e some other value is coming which is not intended. I have a log before

out.collect()

and things are getting printed correctly in that log line.

val userIds = events
      .keyBy(e => e.client)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
      .trigger(
        PurgingTrigger.of(ProcessingTimeTrigger.create())
      ) 
      .process(userIdFetcher)
      .setParallelism(1)
      .uid(userIdFetcher.getClass().getSimpleName())

final class UserIdFetcher(config: UserIdFetcherConfig, failedOutputTag: OutputTag[ReadErrorEvent])
  extends ProcessWindowFunction[WriteSuccessEvent, Diff, String, TimeWindow]
    with Serializable{
  
  elements.groupBy[String](update => update.command.id.get).foreach {
      case (id, updates) =>
       
        retryableProcess(
          id,
          context,
          MongoBulkQuery(
            key,
            "User",
            updates.map(update => update.command.body.identifiers).toList,
            List(UserFields.Id)
          ),
          out
        )
    }
  


  def retryableProcess(
                        id: String,
                        context: Context,
                        query: MongoBulkQuery,
                        out: Collector[Diff],
                        attemptNumber: Int = 0
                      ): Unit = {
    
    mongoClient.client(query.client, query.collection).findUserIDs(query).onComplete {
      case Success(values) =>
       
        values
          .map(user => user.getObjectId(MongoFields.Id).toHexString())
          .grouped(config.maxBatchSize)
          .foreach(userIdBatch => {
            val diff = Diff(query.client, id, userIdBatch.toList)
            out.collect(diff)
          })
    }
  }
}

final case class Diff(
  @JsonProperty("dbName") client: String,
  @JsonProperty("id") collection: String,
  @JsonProperty("add") add: List[String]
)

But in Kafka this value is getting changed randomly of a different

id

which is also getting processed concurrently.

Maybe the kafka topic has multiple partitions? If so you may want to check the partition strategy for the sink (kafka producer client)

But that should not change the value which is getting pushed from out.collect to kafka sink..

Interesting! What data do you see for e.g. in print/log message and what do you see in Kafka message? Examples might help to understand the issue. Also not sure about the

elements.groupBy

call is it called just once from the instantiation of the class?

putting in context ID1 :

Diff(DB1,ID1,List('X'))','timestamp':'2023-04-20T12:39:08.044854445Z'}"

But in Kafka it is getting pushed as (see how ID1 is changing to ID2)

Diff(DB1,ID2,List('X'))','timestamp':'2023-04-20T12:39:08.044854445Z'}"

elements.groupBy

it will be called everytime

ProcessWindowFunction

is called which is every TumblingWindow

if u log the variable

diff

before calling

out.collect(diff)

does it contain ID2 or ID1?

ID1

oh 🙂 what about a log statement after the

.process()

function?

yes, that is also getting printed correctly…

any subsequent functions/operators (like map) that is manipulating the data before pushing to Kafka?

between out.collect and kafka sink we are only doing Byte Array conversion, using objectMapper.writeValueAsBytes()

what serializer do u have configured for kafkaSink?

This is how record is sent to kafka

new ProducerRecord(topic, objectMapper.writeValueAsBytes(key), objectMapper.writeValueAsBytes(value))

From my gut feeling there is something wrong with the serializer. Perhaps you want to move to StringSerializer https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/kafka/#serializer

Ok.. Thanks we will check it out