<commit_before>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
 * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
 * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
 * License. You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.apache.kafka.clients.tools;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class ProducerPerformance {

    private static final long NS_PER_MS = 1000000L;
    private static final long NS_PER_SEC = 1000 * NS_PER_MS;
    private static final long MIN_SLEEP_NS = 2 * NS_PER_MS;

    public static void main(String[] args) throws Exception {
        if (args.length < 4) {
            System.err.println("USAGE: java " + ProducerPerformance.class.getName() +
                               " topic_name num_records record_size target_records_sec [prop_name=prop_value]*");
            System.exit(1);
        }

        /* parse args */
        String topicName = args[0];
        long numRecords = Long.parseLong(args[1]);
        int recordSize = Integer.parseInt(args[2]);
        int throughput = Integer.parseInt(args[3]);

        Properties props = new Properties();
        for (int i = 4; i < args.length; i++) {
            String[] pieces = args[i].split("=");
            if (pieces.length != 2)
                throw new IllegalArgumentException("Invalid property: " + args[i]);
            props.put(pieces[0], pieces[1]);
        }
        KafkaProducer<byte[], byte[]> producer = new KafkaProducer<byte[],byte[]>(props);

        /* setup perf test */
        byte[] payload = new byte[recordSize];
        Arrays.fill(payload, (byte) 1);
        ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>(topicName, payload);
        long sleepTime = NS_PER_SEC / throughput;
        long sleepDeficitNs = 0;
        Stats stats = new Stats(numRecords, 5000);
        long start = System.currentTimeMillis();
        for (int i = 0; i < numRecords; i++) {
            long sendStart = System.currentTimeMillis();
            Callback cb = stats.nextCompletion(sendStart, payload.length, stats);
            producer.send(record, cb);<commit_msg>trivial change to add byte serializer to ProducerPerformance; patched by <NAME><commit_after>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
 * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
 * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
 * License. You may obtain a copy of the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.apache.kafka.clients.tools;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.producer.*;

public class ProducerPerformance {

    private static final long NS_PER_MS = 1000000L;
    private static final long NS_PER_SEC = 1000 * NS_PER_MS;
    private static final long MIN_SLEEP_NS = 2 * NS_PER_MS;

    public static void main(String[] args) throws Exception {
        if (args.length < 4) {
            System.err.println("USAGE: java " + ProducerPerformance.class.getName() +
                               " topic_name num_records record_size target_records_sec [prop_name=prop_value]*");
            System.exit(1);
        }

        /* parse args */
        String topicName = args[0];
        long numRecords = Long.parseLong(args[1]);
        int recordSize = Integer.parseInt(args[2]);
        int throughput = Integer.parseInt(args[3]);

        Properties props = new Properties();
        for (int i = 4; i < args.length; i++) {
            String[] pieces = args[i].split("=");
            if (pieces.length != 2)
                throw new IllegalArgumentException("Invalid property: " + args[i]);
            props.put(pieces[0], pieces[1]);
        }
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        KafkaProducer<byte[], byte[]> producer = new KafkaProducer<byte[],byte[]>(props);

        /* setup perf test */
        byte[] payload = new byte[recordSize];
        Arrays.fill(payload, (byte) 1);
        ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>(topicName, payload);
        long sleepTime = NS_PER_SEC / throughput;
        long sleepDeficitNs = 0;
        Stats stats = new Stats(numRecords, 5000);
        long start = System.currentTimeMillis();
        for (int i = 0; i < numRecords; i++) {
            long sendStart = System.currentTimeMillis();
            Callback cb = stats.nextCompletion(sendStart, payload.length, stats);
            producer.send(record, cb);
{'fraction_non_alphanumeric': 0.10215322984476716, 'fraction_numerical': 0.011016524787180772, 'mean_word_length': 3.08731241473397, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5938859', 'n_tokens_mistral': 1656, 'n_tokens_neox': 1593, 'n_words': 525}