tangyang9464 / kafka-watcher Goto Github PK

public static void main(String[] args) {
    // Initialize the watcher.
    // Use the Kafka properties and topic name as parameter.
    Map<String,Object> producerProps = new HashMap<>();
    producerProps.put("bootstrap.servers", "127.0.0.1:9092");
    producerProps.put("acks", "all");
    producerProps.put("retries", 0);
    producerProps.put("batch.size", 16384);
    producerProps.put("linger.ms", 1);
    producerProps.put("buffer.memory", 33554432);
    producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    Map<String,Object> consumerProps = new HashMap<>();
    consumerProps.put("bootstrap.servers", "127.0.0.1:9092");
    consumerProps.put("group.id", "my-consumer-group-01");
    consumerProps.put("enable.auto.commit", "true");
    consumerProps.put("auto.commit.interval.ms", "1000");
    consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
    String topic="jcasbin-topic";
    KafkaWatcher kafkaWatcher = new KafkaWatcher(producerProps,consumerProps, topic);

    // Initialize the enforcer.
    Enforcer enforcer = new Enforcer("examples/rbac_model.conf", "examples/rbac_policy.csv");

    // Set the watcher for the enforcer.
    enforcer.setWatcher(redisWatcher);

    // Set callback to local example
    Runnable updateCallback = ()->{
        //do something
    };
    kafkaWatcher.setUpdateCallback(updateCallback);

    // Update the policy to test the effect.
    enforcer.savePolicy();
}