<commit_before>package com.codespair.mockstocks.service.kafka.spring.producer;

import com.codespair.mockstocks.config.KafkaConfigProperties;
import com.codespair.mockstocks.model.StockQuote;
import com.fasterxml.jackson.databind.JsonNode;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.connect.json.JsonDeserializer;
import org.apache.kafka.connect.json.JsonSerializer;
import org.springframework.stereotype.Component;

import java.util.Properties;

@Component
public class StockQuoteProducer {

    private KafkaConfigProperties config;
    private Producer<String, JsonNode> kafkaProducer;

    public StockQuoteProducer(KafkaConfigProperties kafkaConfigProperties) {
        this.config = kafkaConfigProperties;
        this.createProducer();
    }

    public void createProducer() {
        kafkaProducer = new KafkaProducer<>(configure());
    }

    public void send(StockQuote stockQuote) {
        kafkaProducer.send();
    }

    private Properties configure() {
        Properties properties = new Properties();

        final Serializer<JsonNode> jsonSerializer = new JsonSerializer();
        final Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();
        final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);

        properties.put("bootstrap.servers", config.getHost());
        properties.put("client.id", config.getStockQuote());
        properties.put("key.serializer", Serdes.String());
        properties.put("value.serializer", jsonSerde);

        return properties;
    }
}
<commit_msg>some work improving kafka producer, moving away from spring kafka<commit_after>package com.codespair.mockstocks.service.kafka.spring.producer;

import com.codespair.mockstocks.config.KafkaConfigProperties;
import com.codespair.mockstocks.model.StockQuote;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.connect.json.JsonDeserializer;
import org.apache.kafka.connect.json.JsonSerializer;
import org.springframework.stereotype.Component;

import java.util.Properties;
import java.util.concurrent.Future;

@Component
public class StockQuoteProducer {

    private KafkaConfigProperties config;
    private Producer<String, JsonNode> kafkaProducer;

    public StockQuoteProducer(KafkaConfigProperties kafkaConfigProperties) {
        this.config = kafkaConfigProperties;
        this.createProducer();
    }

    public void createProducer() {
        kafkaProducer = new KafkaProducer<>(configure());
    }

    public Future<RecordMetadata> send(String key, StockQuote stockQuote) {
        ObjectMapper objectMapper = new ObjectMapper();
        JsonNode stockQuoteJsonNode = objectMapper.convertValue(stockQuote, JsonNode.class);
        return kafkaProducer.send(new ProducerRecord<String, JsonNode>(config.getStockQuote().getTopic(), key,
                stockQuoteJsonNode));
    }

    private Properties configure() {
        Properties properties = new Properties();

        final Serializer<JsonNode> jsonSerializer = new JsonSerializer();
        final Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();
        final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);

        properties.put("bootstrap.servers", config.getHost());
        properties.put("client.id", config.getStockQuote());
        properties.put("key.serializer", Serdes.String());
        properties.put("value.serializer", jsonSerde);

        return properties;
    }
}

{'fraction_non_alphanumeric': 0.09979257893523853, 'fraction_numerical': 0.0, 'mean_word_length': 4.643693107932379, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 3}, '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': '6102334', 'n_tokens_mistral': 1250, 'n_tokens_neox': 1201, 'n_words': 210}