<commit_before>package main

import (
	"flag"
	"fmt"
	"os"
	"os/signal"
	"runtime"
	"strings"
	"sync"
	"syscall"

	"github.com/raintank/worldping-api/pkg/log"
)

var (
	GitHash     = "(none)"
	showVersion = flag.Bool("version", false, "print version string")
	logLevel    = flag.Int("log-level", 2, "log level. 0=TRACE|1=DEBUG|2=INFO|3=WARN|4=ERROR|5=CRITICAL|6=FATAL")

	partitionScheme  = flag.String("partition-scheme", "byOrg", "method used for partitioning metrics. (byOrg|bySeries)")
	compression      = flag.String("compression", "none", "compression: none|gzip|snappy")
	replicateMetrics = flag.Bool("metrics", false, "replicate metrics")
	replicatePersist = flag.Bool("persist", false, "replicate persistMetrics")
	group            = flag.String("group", "mt-replicator", "Kafka consumer group")
	srcTopic         = flag.String("src-topic", "mdm", "topic name on source cluster")
	dstTopic         = flag.String("dst-topic", "mdm", "topic name on destination cluster")
	persistSrcTopic  = flag.String("persist-src-topic", "metricpersist", "metricPersist topic name on source cluster")
	persistDstTopic  = flag.String("persist-dst-topic", "metricpersist", "metricPersist topic name on destination cluster")
	initialOffset    = flag.Int("initial-offset", -2, "initial offset to consume from. (-2=oldest, -1=newest)")
	srcBrokerStr     = flag.String("src-brokers", "localhost:9092", "tcp address of source kafka cluster (may be be given multiple times as a comma-separated list)")
	dstBrokerStr     = flag.String("dst-brokers", "localhost:9092", "tcp address for kafka cluster to consume from (may be be given multiple times as a comma-separated list)")

	wg sync.WaitGroup
)

type topic struct {
	src string
	dst string
}

func main() {
	flag.Usage = func() {
		fmt.Fprintln(os.Stderr, "mt-replicator")
		fmt.Fprintln(os.Stderr)
		fmt.Fprintln(os.Stderr, "Replicates a kafka mdm topic on a given cluster to a topic on another")
		fmt.Fprintf(os.Stderr, "\nFlags:\n\n")
		flag.PrintDefaults()
	}
	flag.Parse()
	log.NewLogger(0, "console", fmt.Sprintf(`{"level": %d, "formatting":false}`, *logLevel))

	if *showVersion {
		fmt.Printf("eventtank (built with %s, git hash %s)\n", runtime.Version(), GitHash)
		return
	}

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	if *group == "" {
		log.Fatal(4, "--group is required")
	}

	if *srcBrokerStr == "" {
		log.Fatal(4, "--src-brokers required")
	}
	if *dstBrokerStr == "" {
		log.Fatal(4, "--dst-brokers required")
	}

	srcBrokers := strings.Split(*srcBrokerStr, ",")
	dstBrokers := strings.Split(*dstBrokerStr, ",")
	wg := new(sync.WaitGroup)

	if *replicateMetrics {

		if *srcTopic == "" {
			log.Fatal(4, "--src-topic is required")
		}

		if *dstTopic == "" {
			log.Fatal(4, "--dst-topic is required")
		}

		consumer, err := NewConsumer(srcBrokers, *group, *srcTopic, *initialOffset)
		if err != nil {
			log.Fatal(4, err.Error())
		}
		publisher, err := NewPublisher(dstBrokers, *dstTopic, *compression, *partitionScheme)
		if err != nil {<commit_msg>clarify src/dst topic<commit_after>package main

import (
	"flag"
	"fmt"
	"os"
	"os/signal"
	"runtime"
	"strings"
	"sync"
	"syscall"

	"github.com/raintank/worldping-api/pkg/log"
)

var (
	GitHash     = "(none)"
	showVersion = flag.Bool("version", false, "print version string")
	logLevel    = flag.Int("log-level", 2, "log level. 0=TRACE|1=DEBUG|2=INFO|3=WARN|4=ERROR|5=CRITICAL|6=FATAL")

	partitionScheme  = flag.String("partition-scheme", "byOrg", "method used for partitioning metrics. (byOrg|bySeries)")
	compression      = flag.String("compression", "none", "compression: none|gzip|snappy")
	replicateMetrics = flag.Bool("metrics", false, "replicate metrics")
	replicatePersist = flag.Bool("persist", false, "replicate persistMetrics")
	group            = flag.String("group", "mt-replicator", "Kafka consumer group")
	metricSrcTopic   = flag.String("metric-src-topic", "mdm", "metrics topic name on source cluster")
	metricDstTopic   = flag.String("metric-dst-topic", "mdm", "metrics topic name on destination cluster")
	persistSrcTopic  = flag.String("persist-src-topic", "metricpersist", "metricPersist topic name on source cluster")
	persistDstTopic  = flag.String("persist-dst-topic", "metricpersist", "metricPersist topic name on destination cluster")
	initialOffset    = flag.Int("initial-offset", -2, "initial offset to consume from. (-2=oldest, -1=newest)")
	srcBrokerStr     = flag.String("src-brokers", "localhost:9092", "tcp address of source kafka cluster (may be be given multiple times as a comma-separated list)")
	dstBrokerStr     = flag.String("dst-brokers", "localhost:9092", "tcp address for kafka cluster to consume from (may be be given multiple times as a comma-separated list)")

	wg sync.WaitGroup
)

type topic struct {
	src string
	dst string
}

func main() {
	flag.Usage = func() {
		fmt.Fprintln(os.Stderr, "mt-replicator")
		fmt.Fprintln(os.Stderr)
		fmt.Fprintln(os.Stderr, "Replicates a kafka mdm topic on a given cluster to a topic on another")
		fmt.Fprintf(os.Stderr, "\nFlags:\n\n")
		flag.PrintDefaults()
	}
	flag.Parse()
	log.NewLogger(0, "console", fmt.Sprintf(`{"level": %d, "formatting":false}`, *logLevel))

	if *showVersion {
		fmt.Printf("eventtank (built with %s, git hash %s)\n", runtime.Version(), GitHash)
		return
	}

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

	if *group == "" {
		log.Fatal(4, "--group is required")
	}

	if *srcBrokerStr == "" {
		log.Fatal(4, "--src-brokers required")
	}
	if *dstBrokerStr == "" {
		log.Fatal(4, "--dst-brokers required")
	}

	srcBrokers := strings.Split(*srcBrokerStr, ",")
	dstBrokers := strings.Split(*dstBrokerStr, ",")
	wg := new(sync.WaitGroup)

	if *replicateMetrics {

		if *metricSrcTopic == "" {
			log.Fatal(4, "--metric-src-topic is required")
		}

		if *metricDstTopic == "" {
			log.Fatal(4, "--metric-dst-topic is required")
		}

		consumer, err := NewConsumer(srcBrokers, *group, *metricSrcTopic, *initialOffset)
		if err != nil {
			log.Fatal(4, err.Error())
		}
		publisher, err := NewPublisher(dstBrokers, *metricDstTopic, *compression, *partitionScheme)
		if err != nil {
{'fraction_non_alphanumeric': 0.15934775589279948, 'fraction_numerical': 0.00871811430416532, 'mean_word_length': 4.462962962962963, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, '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': '5184267', 'n_tokens_mistral': 2278, 'n_tokens_neox': 2131, 'n_words': 518}