<commit_before>#!/usr/bin/perl

use strict;



my $port_prefix = $ENV{C4PORT_PREFIX} || 8000;
my $http_port = $port_prefix+67;
my $sse_port = $port_prefix+68;
my $zoo_port = $port_prefix+81;
my $kafka_port = $port_prefix+92;
my $build_dir = "./client/build/test";
my $inbox_prefix = '';
my $kafka_version = "0.10.2.0";
my $kafka = "kafka_2.11-$kafka_version";
my $curl_test = "curl http://127.0.0.1:$http_port/abc";
my $bootstrap_server = "127.0.0.1:$kafka_port";


sub sy{ print join(" ",@_),"\n"; system @_ and die $?; }
sub syf{ my $res = scalar `$_[0]`; print "$_[0]\n$res"; $res }

my $put_text = sub{
    my($fn,$content)=@_;
    open FF,">:encoding(UTF-8)",$fn and print FF $content and close FF or die "put_text($!)($fn)";
};

my @tasks;

push @tasks, ["setup_sbt", sub{
    (-e $_ or mkdir $_) and chdir $_ or die for "tmp";
    my $sbta = "sbt-0.13.13.tgz";
    if(!-e $sbta){
        sy("wget https://dl.bintray.com/sbt/native-packages/sbt/0.13.13/$sbta");
        sy("tar -xzf $sbta");
        sy("./sbt-launcher-packaging-0.13.13/bin/sbt update")
    }

    #my $nodea = "node-v6.10.0-linux-x64.tar.xz";
    #if(!-e $nodea){
    #    sy("wget https://nodejs.org/dist/v6.10.0/$nodea");
    #    sy("tar -xJf $nodea");
    #}
    #print qq{export PATH=tmp/sbt-launcher-packaging-0.13.13/bin:tmp/node-v6.10.0-linux-x64/bin:\$PATH\n};
    print qq{add to .bashrc or so:\nexport PATH=tmp/sbt-launcher-packaging-0.13.13/bin:\$PATH\n};
}];
push @tasks, ["setup_kafka", sub{
    (-e $_ or mkdir $_) and chdir $_ or die for "tmp";
    if (!-e $kafka) {
        sy("wget http://www-eu.apache.org/dist/kafka/$kafka_version/$kafka.tgz");
        sy("tar -xzf $kafka.tgz")
    }
}];


push @tasks, ["es_examples", sub{
    sy("sbt 'c4actor-base-examples/run-main ee.cone.c4actor.ProtoAdapterTest' ");
    sy("sbt 'c4actor-base-examples/run-main ee.cone.c4actor.AssemblerTest' ");
}];
push @tasks, ["not_effective_join_bench", sub{
    sy("sbt 'c4actor-base-examples/run-main ee.cone.c4actor.NotEffectiveAssemblerTest' ");
}];

my $inbox_configure = sub{
    my $kafka_topics = "tmp/$kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:$zoo_port --topic .inbox";
    sy("$kafka_topics --create --partitions 1 --replication-factor 1")
        if 0 > index syf("$kafka_topics --list"), ".inbox";
    my $kafka_configs = "tmp/$kafka/bin/kafka-configs.sh --zookeeper 127.0.0.1:$zoo_port --entity-type topics ";
    my $infinite_lag = "min.compaction.lag.ms=9223372036854775807";
    my $compression = "compression.type=producer";
    sy("$kafka_configs --alter --entity-name .inbox --add-config $infinite_lag,$compression");
    die if 0 > index syf("$kafka_configs --describe --entity-name .inbox"),$infinite_lag;
};

push @tasks, ["start_kafka", sub{
    &$put_text("tmp/zookeeper.properties","dataDir=db4/zookeeper\nclientPort=$zoo_port\n");
    &$put_text("tmp/server.properties",join "\n",
        "listeners=PLAINTEXT://$bootstrap_server",
        "log.dirs=db4/kafka-logs",
        "zookeeper.connect=127.0.0.1:$zoo_port",
        "log.cleanup.policy=compact",
        "log.segment.bytes=104857600", #active segment is not compacting, so we reduce it
        "log.cleaner.delete.retention.ms=3600000", #1h
        "log.roll.hours=1", #delete-s will be triggered to remove?
        "compression.type=uncompressed", #probably better compaction for .state topics
        "message.max.bytes=3000000" #seems to be compressed
    );
    sy("tmp/$kafka/bin/zookeeper-server-start.sh -daemon tmp/zookeeper.properties");
    sy("tmp/$kafka/bin/kafka-server-start.sh -daemon tmp/server.properties");
    sy("jps");
    &$inbox_configure();
}];
push @tasks, ["stop_kafka", sub{
    sy("tmp/$kafka/bin/kafka-server-stop.sh")
    sy("tmp/$kafka/bin/zookeeper-server-stop.sh")
}];

push @tasks, ["inbox_configure", sub{&$inbox_configure()}];

push @tasks, ["inbox_log_tail", sub{
    sy("tmp/$kafka/bin/kafka-console-consumer.sh --bootstrap-server $bootstrap_server --topic $inbox_prefix.inbox.log")
}];
push @tasks, ["inbox_test", sub{
    sy("tmp/$kafka/bin/kafka-verifiable-consumer.sh --broker-list $bootstrap_server --topic $inbox_prefix.inbox --group-id dummy-".rand())
}];

=sk
push @tasks, ["inbox_copy", sub{
    my $from = $ENV{C4COPY_FROM} || die "C4COPY_FROM required";
    &$put_text("tmp/copy.consumer.properties",join "\n",
        "group.id=dummy-".rand(),
        "bootstrap.servers=$from",
        #"enable.auto.commit=false"
    );
    &$put_text("tmp/copy.producer.properties",join "\n",
        "bootstrap.servers=$bootstrap_server",
        "compression.type=lz4",
        "max.request.size=10000000",
        #"linger.ms=1000",
        "batch.size=1000",
    );
    sy("tmp/$kafka/bin/kafka-mirror-maker.sh"
        ." --consumer.config tmp/copy.consumer.properties"
        ." --producer.config tmp/copy.producer.properties"<commit_msg>zk stop<commit_after>#!/usr/bin/perl

use strict;



my $port_prefix = $ENV{C4PORT_PREFIX} || 8000;
my $http_port = $port_prefix+67;
my $sse_port = $port_prefix+68;
my $zoo_port = $port_prefix+81;
my $kafka_port = $port_prefix+92;
my $build_dir = "./client/build/test";
my $inbox_prefix = '';
my $kafka_version = "0.10.2.0";
my $kafka = "kafka_2.11-$kafka_version";
my $curl_test = "curl http://127.0.0.1:$http_port/abc";
my $bootstrap_server = "127.0.0.1:$kafka_port";


sub sy{ print join(" ",@_),"\n"; system @_ and die $?; }
sub syf{ my $res = scalar `$_[0]`; print "$_[0]\n$res"; $res }

my $put_text = sub{
    my($fn,$content)=@_;
    open FF,">:encoding(UTF-8)",$fn and print FF $content and close FF or die "put_text($!)($fn)";
};

my @tasks;

push @tasks, ["setup_sbt", sub{
    (-e $_ or mkdir $_) and chdir $_ or die for "tmp";
    my $sbta = "sbt-0.13.13.tgz";
    if(!-e $sbta){
        sy("wget https://dl.bintray.com/sbt/native-packages/sbt/0.13.13/$sbta");
        sy("tar -xzf $sbta");
        sy("./sbt-launcher-packaging-0.13.13/bin/sbt update")
    }

    #my $nodea = "node-v6.10.0-linux-x64.tar.xz";
    #if(!-e $nodea){
    #    sy("wget https://nodejs.org/dist/v6.10.0/$nodea");
    #    sy("tar -xJf $nodea");
    #}
    #print qq{export PATH=tmp/sbt-launcher-packaging-0.13.13/bin:tmp/node-v6.10.0-linux-x64/bin:\$PATH\n};
    print qq{add to .bashrc or so:\nexport PATH=tmp/sbt-launcher-packaging-0.13.13/bin:\$PATH\n};
}];
push @tasks, ["setup_kafka", sub{
    (-e $_ or mkdir $_) and chdir $_ or die for "tmp";
    if (!-e $kafka) {
        sy("wget http://www-eu.apache.org/dist/kafka/$kafka_version/$kafka.tgz");
        sy("tar -xzf $kafka.tgz")
    }
}];


push @tasks, ["es_examples", sub{
    sy("sbt 'c4actor-base-examples/run-main ee.cone.c4actor.ProtoAdapterTest' ");
    sy("sbt 'c4actor-base-examples/run-main ee.cone.c4actor.AssemblerTest' ");
}];
push @tasks, ["not_effective_join_bench", sub{
    sy("sbt 'c4actor-base-examples/run-main ee.cone.c4actor.NotEffectiveAssemblerTest' ");
}];

my $inbox_configure = sub{
    my $kafka_topics = "tmp/$kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:$zoo_port --topic .inbox";
    sy("$kafka_topics --create --partitions 1 --replication-factor 1")
        if 0 > index syf("$kafka_topics --list"), ".inbox";
    my $kafka_configs = "tmp/$kafka/bin/kafka-configs.sh --zookeeper 127.0.0.1:$zoo_port --entity-type topics ";
    my $infinite_lag = "min.compaction.lag.ms=9223372036854775807";
    my $compression = "compression.type=producer";
    sy("$kafka_configs --alter --entity-name .inbox --add-config $infinite_lag,$compression");
    die if 0 > index syf("$kafka_configs --describe --entity-name .inbox"),$infinite_lag;
};

push @tasks, ["start_kafka", sub{
    &$put_text("tmp/zookeeper.properties","dataDir=db4/zookeeper\nclientPort=$zoo_port\n");
    &$put_text("tmp/server.properties",join "\n",
        "listeners=PLAINTEXT://$bootstrap_server",
        "log.dirs=db4/kafka-logs",
        "zookeeper.connect=127.0.0.1:$zoo_port",
        "log.cleanup.policy=compact",
        "log.segment.bytes=104857600", #active segment is not compacting, so we reduce it
        "log.cleaner.delete.retention.ms=3600000", #1h
        "log.roll.hours=1", #delete-s will be triggered to remove?
        "compression.type=uncompressed", #probably better compaction for .state topics
        "message.max.bytes=3000000" #seems to be compressed
    );
    sy("tmp/$kafka/bin/zookeeper-server-start.sh -daemon tmp/zookeeper.properties");
    sy("tmp/$kafka/bin/kafka-server-start.sh -daemon tmp/server.properties");
    sy("jps");
    &$inbox_configure();
}];
push @tasks, ["stop_kafka", sub{
    sy("tmp/$kafka/bin/kafka-server-stop.sh");
    sy("tmp/$kafka/bin/zookeeper-server-stop.sh");
}];

push @tasks, ["inbox_configure", sub{&$inbox_configure()}];

push @tasks, ["inbox_log_tail", sub{
    sy("tmp/$kafka/bin/kafka-console-consumer.sh --bootstrap-server $bootstrap_server --topic $inbox_prefix.inbox.log")
}];
push @tasks, ["inbox_test", sub{
    sy("tmp/$kafka/bin/kafka-verifiable-consumer.sh --broker-list $bootstrap_server --topic $inbox_prefix.inbox --group-id dummy-".rand())
}];

=sk
push @tasks, ["inbox_copy", sub{
    my $from = $ENV{C4COPY_FROM} || die "C4COPY_FROM required";
    &$put_text("tmp/copy.consumer.properties",join "\n",
        "group.id=dummy-".rand(),
        "bootstrap.servers=$from",
        #"enable.auto.commit=false"
    );
    &$put_text("tmp/copy.producer.properties",join "\n",
        "bootstrap.servers=$bootstrap_server",
        "compression.type=lz4",
        "max.request.size=10000000",
        #"linger.ms=1000",
        "batch.size=1000",
    );
    sy("tmp/$kafka/bin/kafka-mirror-maker.sh"
        ." --consumer.config tmp/copy.consumer.properties"
        ." --producer.config tmp/copy.producer.properties"
{'fraction_non_alphanumeric': 0.1959933911606774, 'fraction_numerical': 0.03510945890128046, 'mean_word_length': 3.45902394106814, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 12, 'substrings_counts': 2, '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': '2411723', 'n_tokens_mistral': 4022, 'n_tokens_neox': 3765, 'n_words': 624}