spring.cloud.stream:
  default:
    producer:
      useNativeEncoding: true
      partition-key-expression: headers['kafka_messageKey'].id
    consumer.useNativeDecoding: true
  function.autodetect: false
  bindings:
    testEventSource-out-0:
      destination: test-event
  kafka:
    binder:
      auto-create-topics: false
      producer-properties:
        key.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
        value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
        key.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
        value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

void sendEvent_whenUsingStreamBridge_shouldRespectPartitioningScheme() {
        final int numberOfPartitions = 6;

        final Message<KafkaNull> outgoingMessage = MessageBuilder.withPayload(KafkaNull.INSTANCE)
                .setHeader(KafkaHeaders.KEY, TestEventKey.newBuilder().setId(1001L).build())
                .build();

        streamBridge.send("testEventSource-out-0", outgoingMessage);

        final ConsumerRecords<Object, Object> records = testConsumer.poll(Duration.ofSeconds(5L));
        assertThat(records.iterator().next().partition()).isEqualTo(eventId % numberOfPartitions);
    }