Changing Headers definition from byte to int8 for KafkaRecord

[milovacb]

Description of changes:

The Headers field needs to reflect value range of byte data type in the Java library.

The unit test example is consistent with public docs here: https://docs.aws.amazon.com/lambda/latest/dg/with-kafka.html#smaa-sample-event

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov-commenter]

Codecov Report

Merging #506 (a597bf6) into main (81d63bd) will increase coverage by 1.25%.
The diff coverage is 81.25%.

@@            Coverage Diff             @@
##             main     #506      +/-   ##
==========================================
+ Coverage   71.38%   72.63%   +1.25%     
==========================================
  Files          22       26       +4     
  Lines        1307     1407     +100     
==========================================
+ Hits          933     1022      +89     
- Misses        306      314       +8     
- Partials       68       71       +3     

Impacted Files	Coverage Δ
events/kafka.go	81.25% <81.25%> (ø)

... and 3 files with indirect coverage changes

[bmoffatt]

More relationships:

• Similar fix was recently made for the Lambda .NET project Updated the KafkaEvent header(s) value to use the correct array type of sbyte values, so that negative values could be handled. aws-lambda-dotnet#1476
• Fixes: [BUG] Incorrect unmarshalling self-hosted Kafka event #422

[bmoffatt]

So, since this field was never functional since being added in #385, my gut reaction is that I'd be OK merging this as-is and calling it a bug fix, rather than a semver breaking change.

However, aws/aws-lambda-dotnet#1476 (comment), makes me think it'd be possible to keep strict back-compatability by adding a custom Marshal/Unmarshal JSON to this struct, so that the field type may stay as []byte. This might also be more convienient to work with for consumers, as []int8 isn't really a common type to be playing around with in application code. Custom Marshaling isn't abnormal for events in this library (ex: recently merged a custom marshaling in #502!)

@milovacb WDYT? Does it look feasible to do a bit of extra work to allow Go consumers to consume the kafka header value as []byte, rather than having to convert it themselves? Or is there some Kafka weirdness (I'm far from an expert) where this needs to be handled as signed bytes within application code too?

[milovacb]

I researched this a little bit in more detail. Official Kafka Go library does use byte in its implementation. However, this is still going to be used inside of a Lambda function invoked by ESM (Event Source Mapping) and negative values may arrive.

It comes down to customer experience - should this library match the experience of Kafka Go library or should it match Lambda experience? My experience so far was that Lambda experience was more important for customers, but I don't think I have enough data for this to make that decision and this being an open source library I guess we arrive to conclusions in a different way.

I think users can make an appropriate conversion by themselves if needed, but if it is a blocker I can make conversion to byte by adding 256 to all negative values (thus all negative values with 1 as the highest bit will be translated to 128-255 range)?

[bmoffatt]

I wouldn't expect a consumer to ever interact with the "negative numbers" directly. The producer, regardless of how the particular producer chooses to pretty-print it's bytes, chose to write bytes. So the consumer should get bytes too!

I think something like this would result in a more ergonomic event type

type KafkaFooBar struct {
   //...
   Headers map[string]JSONNumberBytes
   //....
}

// JSONNumberBytes is bytes that are json encoded/decoded as numbers in the range [-128, 127]
type JSONNumberBytes []byte

func (b JSONNumberBytes) MarshalJSON() ([]byte, error) {
	n := make([]int8, len(b))
	for i, v := range b {
		n[i] = int8(v)
	}
	return json.Marshal(n)
}

func (b *JSONNumberBytes) UnmarshalJSON(p []byte) error {
	var n []int8
	if err := json.Unmarshal(p, &n); err != nil {
		return err
	}
	*b = make(JSONNumberBytes, len(n))
	for i, v := range n {
		(*b)[i] = byte(v)
	}
	return nil
}

https://go.dev/play/p/7_10l_AIsxF

[milovacb]

OK, keeping bytes then. Test updated.

[bmoffatt]

Thanks!