build-on-aws / prioritizing-event-processing-with-apache-kafka Goto Github PK

    public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
        // With the introduction of KIP-480 to enhance record production
        // throughput Kafka's API calls the partition() method twice resulting
        // in partitions being skipped. More information about this here:
        // https://issues.apache.org/jira/browse/KAFKA-9965
        // The temporary solution is to use the callback method 'onNewBatch'
        // to decrease the counter to stabilize the round-robin logic.
        String bucketName = lastBucket.get();
        Bucket bucket = buckets.get(bucketName);
        if (bucket != null) {
            bucket.decrementCounter();
        }
        lastBucket.remove();
    }

java -cp target/bucket-priority-pattern-1.0.0-jar-with-dependencies.jar blog.buildon.aws.streaming.kafka.BucketBasedProducer

java -cp target/bucket-priority-pattern-1.0.0-jar-with-dependencies.jar blog.buildon.aws.streaming.kafka.BucketBasedConsumer Platinum 5