Having a problem to use Int64SerDe

[jdang67]

Hi,
I got a topic that uses the key as Int64(long). My consumer app does not have any problem to ingest the data.
However, when I try to use Streamiz.Kafka.Net to process my topic, the value of the key is not the same as the input key.
Does anyone see any problem with my code or did I discover a bug?

Expected Value Actual Value
1 72057594037927936
2 144115188075855872

            var config = new StreamConfig<Int64SerDes,JsonSerDes<TeamInfo>>();
            config.ApplicationId = "test-app-teaminfo";
            config.BootstrapServers = "localhost:9092";
            config.AutoOffsetReset = Confluent.Kafka.AutoOffsetReset.Earliest;
            config.StateDir = @"C:/Temp/kafka-stream/";
            var keys  = new List<long>();
            var builder = new StreamBuilder();
            builder.Stream<long, TeamInfo>("TeamInfo")
                .Foreach((k, v) =>
                {
                    keys.Add(k);
                });

            var topology = builder.Build();
            var stream = new KafkaStream(topology, config);
            stream.Start();

[LGouellec]

Hi @jdang67,

What is the nature of the application which produce inside your source topic ? A Java producer or .net producer ?

Can you share the snippet of code of your producer, mainly the serializer please ?

It seems an endianness issue.

[jdang67]

@LGouellec,

Using .Net producer. At first, I used method 1( no serializer for the key), and both KafkaUI and my consumer apps worked just fine. Ran into a problem with Streamiz.Kafka.Net, I tried to use Serializers.Int64; and I have the same problem.

Debugging Streamiz.Kafka.Net app,
the actual value of the key should be { 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0 };
but I see this { 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1 };

Let me know if you see anything wrong with my code snippets

thanks,

Jdang

var config = KafkaUtilities.GetProducerConfig(_bootstrapServers);

  var valueSerializer = new JsonSerializer < TeamInfo > ();

  // Method 1, not using keySerializer 
  using(var producer = new ProducerBuilder < long, TeamInfo > (config).SetValueSerializer(valueSerializer).Build()) {
    foreach(var team in teams) {
      var message = new Message < long,
        TeamInfo > {
          Key = team.TeamId,
          Value = team
        };

      var deliveryReport = await producer.ProduceAsync(topic, message);
      KafkaUtilities.OutputMessage < long, TeamInfo > (deliveryReport);
    }
  }

  // Method 2, using  keySerializer  as Int64
  var keySerializer = Serializers.Int64;
  using(var producer = new ProducerBuilder < long, TeamInfo > (config).SetValueSerializer(valueSerializer).SetKeySerializer(keySerializer).Build()) {
    foreach(var team in teams) {
      var message = new Message < long,
        TeamInfo > {
          Key = team.TeamId,
          Value = team
        };

      var deliveryReport = await producer.ProduceAsync(topic, message);
      KafkaUtilities.OutputMessage < long, TeamInfo > (deliveryReport);
    }
  }

[LGouellec]

Hi @jdang67,

The Serializers.Int64 serialize your integer as a big endian byte array (https://github.com/confluentinc/confluent-kafka-dotnet/blob/a67bd6c06b7eef4293e6476d9ff6f3e93f0e4cd9/src/Confluent.Kafka/Serializers.cs#L62).

The Int64SerDes use internally BitConverter. The order of bytes used depends on whether the computer architecture is little-endian or big-endian. So typically, it's highly likely that your computer is little-endian, but your producer serializer use big-endian, that's why the actual value vs expected value seems reverse.

the actual value of the key should be { 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0 };
but I see this { 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x1 };

As a workaround, you can implement your own SerDes to enforce a big endian deserializer.

    public class BigEndianInt64SerDes : AbstractSerDes<long>
    {
        /// <summary>
        /// Deserialize a record value from a big endian byte array into <see cref="Int64"/> value
        /// </summary>
        /// <param name="data">serialized bytes in big endian.</param>
        /// <param name="context">serialization context</param>
        /// <returns>deserialized <see cref="Int64"/> using data; may be null</returns>
        public override long Deserialize(byte[] data, SerializationContext context)
        {
            return System.Buffers.Binary.BinaryPrimitives.ReadInt64BigEndian(data);
        }

        /// <summary>
        /// Convert long <code>data</code> into a big endian byte array
        /// </summary>
        /// <param name="data"><see cref="Int64"/> data</param>
        /// <param name="context">serialization context</param>
        /// <returns>serialized bytes in big endian</returns>
        public override byte[] Serialize(long data, SerializationContext context)
        {
            Span<byte> span = new Span<byte>();
            System.Buffers.Binary.BinaryPrimitives.WriteInt64BigEndian(span, data);
            return span.ToArray();
        }
    }

@jdang67 Can you test it and said me if it's fixed ?

[jdang67]

@LGouellec,

I have not tried to produce the data again with your new Serde. To make your code work, see the code , I need to make a minor change
otherwise, it throws an exception.

public class BigEndianInt64SerDes : AbstractSerDes<long>
    {
        /// <summary>
        /// Deserialize a record value from a big endian byte array into <see cref="Int64"/> value
        /// </summary>
        /// <param name="data">serialized bytes in big endian.</param>
        /// <param name="context">serialization context</param>
        /// <returns>deserialized <see cref="Int64"/> using data; may be null</returns>
        public override long Deserialize(byte[] data, SerializationContext context)
        {
            return System.Buffers.Binary.BinaryPrimitives.ReadInt64BigEndian(data);
        }

        /// <summary>
        /// Convert long <code>data</code> into a big endian byte array
        /// </summary>
        /// <param name="data"><see cref="Int64"/> data</param>
        /// <param name="context">serialization context</param>
        /// <returns>serialized bytes in big endian</returns>
        public override byte[] Serialize(long data, SerializationContext context)
        {
            Span<byte> span = new Span<byte>();
            // System.Buffers.Binary.BinaryPrimitives.WriteInt64BigEndian(span, data);

             return BitConverter.GetBytes(data);
            
            //return span.ToArray();
        }
    }

[LGouellec]

With this serdes, you deserialize with a big endian policy and publish a sink record with a little endian policy.

[jdang67]

I'm not sure why we need to have another Sedes to deal with this problem while other systems do not have to. I believe we might be able to add a few lines of code in your Int64Serde to fix this problem since we are dealing with long, not directly with big or little Endian.

[LGouellec]

@jdang67 Sure it can be an option, but as a workaround you can implement your own serdes.
SerDes is explicitly public to allow the possibility to create your own serdes based on your use case.

Today the long .net serializer and the long java serializer as well serialize long with a big endian strategy, so if you would like to consume this topic with a Streamiz application, the Int64SerDes can't deserialize the payload correctly.

Feel free to create an GH issue to track the lack.