<commit_before>package com.github.gquintana.kafka;

import kafka.Kafka;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.utils.KafkaTestUtils;

import java.util.Collections;

import static com.github.gquintana.kafka.MessageServiceHelper.sendAndConsume;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.kafka.test.assertj.KafkaConditions.value;

//tag::start[]
public class SpringMessageServiceIT {
    private static final String TOPIC = "spring";
    @ClassRule //<1>
    public static EmbeddedKafkaRule kafka = new EmbeddedKafkaRule(1,
            false, TOPIC);
    @Test
    public void testSendAndConsume() throws Exception {
        sendAndConsume(kafka.getEmbeddedKafka().getBrokersAsString(), TOPIC);
    }
    //end::start[]

    @Test
    public void testAPI() throws Exception {

        try(Consumer<Integer, String> consumer = new KafkaConsumer<Integer, String>( //<1>
                KafkaTestUtils.consumerProps("spring_group", "true", kafka.getEmbeddedKafka()))) {
            KafkaTemplate<Integer, String> template = new KafkaTemplate<>( //<2>
                    new DefaultKafkaProducerFactory<>(
                            KafkaTestUtils.producerProps(kafka.getEmbeddedKafka())));
            consumer.subscribe(Collections.singleton(TOPIC));

            template.send(TOPIC, "one");
            template.send(TOPIC, "two");

            ConsumerRecords<Integer, String> records = KafkaTestUtils.getRecords(consumer); //<3>
            assertThat(records).are(value("one")); //<4>
        }

    }

}<commit_msg>Add Kafka Spring IT tags<commit_after>package com.github.gquintana.kafka;

import kafka.Kafka;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.utils.KafkaTestUtils;

import java.util.Collections;

import static com.github.gquintana.kafka.MessageServiceHelper.sendAndConsume;
import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.kafka.test.assertj.KafkaConditions.value;

//tag::start[]
public class SpringMessageServiceIT {
    private static final String TOPIC = "spring";
    @ClassRule //<1>
    public static EmbeddedKafkaRule kafka = new EmbeddedKafkaRule(1,
            false, TOPIC);
    @Test
    public void testSendAndConsume() throws Exception {
        sendAndConsume(kafka.getEmbeddedKafka().getBrokersAsString(), TOPIC);
    }
    //end::start[]

    @Test
    public void testAPI() throws Exception {
        //tag::api[]
        try(Consumer<Integer, String> consumer = new KafkaConsumer<Integer, String>( //<1>
                KafkaTestUtils.consumerProps("spring_group", "true", kafka.getEmbeddedKafka()))) {
            KafkaTemplate<Integer, String> template = new KafkaTemplate<>( //<2>
                    new DefaultKafkaProducerFactory<>(
                            KafkaTestUtils.producerProps(kafka.getEmbeddedKafka())));
            consumer.subscribe(Collections.singleton(TOPIC));

            template.send(TOPIC, "one");
            template.send(TOPIC, "two");

            ConsumerRecords<Integer, String> records = KafkaTestUtils.getRecords(consumer); //<3>
            assertThat(records).are(value("one")); //<4
            //tag::api[]
        }

    }

}
{'fraction_non_alphanumeric': 0.114413776834159, 'fraction_numerical': 0.0028308563340410475, 'mean_word_length': 3.742729306487696, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5049747', 'n_tokens_mistral': 1262, 'n_tokens_neox': 1227, 'n_words': 212}