| /** |
| * 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.pulsar.common.policies.data; |
| |
| import com.google.common.base.MoreObjects; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Sets; |
| |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.Set; |
| import javax.ws.rs.core.Response.Status; |
| |
| import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; |
| import org.apache.pulsar.common.util.RestException; |
| |
| /** |
| * Definition of Pulsar policies. |
| */ |
| public class Policies { |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public final AuthPolicies auth_policies = new AuthPolicies(); |
| @SuppressWarnings("checkstyle:MemberName") |
| public Set<String> replication_clusters = Sets.newHashSet(); |
| public BundlesData bundles; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Map<BacklogQuota.BacklogQuotaType, BacklogQuota> backlog_quota_map = Maps.newHashMap(); |
| @Deprecated |
| public Map<String, DispatchRate> clusterDispatchRate = Maps.newHashMap(); |
| public Map<String, DispatchRate> topicDispatchRate = Maps.newHashMap(); |
| public Map<String, DispatchRate> subscriptionDispatchRate = Maps.newHashMap(); |
| public Map<String, DispatchRate> replicatorDispatchRate = Maps.newHashMap(); |
| public Map<String, SubscribeRate> clusterSubscribeRate = Maps.newHashMap(); |
| public PersistencePolicies persistence = null; |
| |
| // If set, it will override the broker settings for enabling deduplication |
| public Boolean deduplicationEnabled = null; |
| // If set, it will override the broker settings for allowing auto topic creation |
| public AutoTopicCreationOverride autoTopicCreationOverride = null; |
| // If set, it will override the broker settings for allowing auto subscription creation |
| public AutoSubscriptionCreationOverride autoSubscriptionCreationOverride = null; |
| public Map<String, PublishRate> publishMaxMessageRate = Maps.newHashMap(); |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public Map<String, Integer> latency_stats_sample_rate = Maps.newHashMap(); |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer message_ttl_in_seconds = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public int subscription_expiration_time_minutes = 0; |
| @SuppressWarnings("checkstyle:MemberName") |
| public RetentionPolicies retention_policies = null; |
| public boolean deleted = false; |
| public static final String FIRST_BOUNDARY = "0x00000000"; |
| public static final String LAST_BOUNDARY = "0xffffffff"; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public boolean encryption_required = false; |
| @SuppressWarnings("checkstyle:MemberName") |
| public DelayedDeliveryPolicies delayed_delivery_policies = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public InactiveTopicPolicies inactive_topic_policies = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public SubscriptionAuthMode subscription_auth_mode = SubscriptionAuthMode.None; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer max_producers_per_topic = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer max_consumers_per_topic = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer max_consumers_per_subscription = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer max_unacked_messages_per_consumer = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer max_unacked_messages_per_subscription = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer max_subscriptions_per_topic = null; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public Long compaction_threshold = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public long offload_threshold = -1; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Long offload_deletion_lag_ms = null; |
| @SuppressWarnings("checkstyle:MemberName") |
| public Integer max_topics_per_namespace = null; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| @Deprecated |
| public SchemaAutoUpdateCompatibilityStrategy schema_auto_update_compatibility_strategy = |
| SchemaAutoUpdateCompatibilityStrategy.Full; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public SchemaCompatibilityStrategy schema_compatibility_strategy = SchemaCompatibilityStrategy.UNDEFINED; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public boolean is_allow_auto_update_schema = true; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public boolean schema_validation_enforced = false; |
| |
| @SuppressWarnings("checkstyle:MemberName") |
| public OffloadPolicies offload_policies = null; |
| |
| public Integer deduplicationSnapshotIntervalSeconds = null; |
| |
| public Set<SubType> subscription_types_enabled = Sets.newHashSet(); |
| |
| @Override |
| public int hashCode() { |
| return Objects.hash(auth_policies, replication_clusters, |
| backlog_quota_map, publishMaxMessageRate, clusterDispatchRate, |
| topicDispatchRate, subscriptionDispatchRate, replicatorDispatchRate, |
| clusterSubscribeRate, deduplicationEnabled, autoTopicCreationOverride, |
| autoSubscriptionCreationOverride, persistence, |
| bundles, latency_stats_sample_rate, |
| message_ttl_in_seconds, subscription_expiration_time_minutes, retention_policies, |
| encryption_required, delayed_delivery_policies, inactive_topic_policies, |
| subscription_auth_mode, |
| max_producers_per_topic, |
| max_consumers_per_topic, max_consumers_per_subscription, |
| max_unacked_messages_per_consumer, max_unacked_messages_per_subscription, |
| compaction_threshold, offload_threshold, |
| offload_deletion_lag_ms, |
| schema_auto_update_compatibility_strategy, |
| schema_validation_enforced, |
| schema_compatibility_strategy, |
| is_allow_auto_update_schema, |
| offload_policies, |
| subscription_types_enabled); |
| } |
| |
| @Override |
| public boolean equals(Object obj) { |
| if (obj instanceof Policies) { |
| Policies other = (Policies) obj; |
| return Objects.equals(auth_policies, other.auth_policies) |
| && Objects.equals(replication_clusters, other.replication_clusters) |
| && Objects.equals(backlog_quota_map, other.backlog_quota_map) |
| && Objects.equals(clusterDispatchRate, other.clusterDispatchRate) |
| && Objects.equals(topicDispatchRate, other.topicDispatchRate) |
| && Objects.equals(subscriptionDispatchRate, other.subscriptionDispatchRate) |
| && Objects.equals(replicatorDispatchRate, other.replicatorDispatchRate) |
| && Objects.equals(clusterSubscribeRate, other.clusterSubscribeRate) |
| && Objects.equals(publishMaxMessageRate, other.publishMaxMessageRate) |
| && Objects.equals(deduplicationEnabled, other.deduplicationEnabled) |
| && Objects.equals(autoTopicCreationOverride, other.autoTopicCreationOverride) |
| && Objects.equals(autoSubscriptionCreationOverride, other.autoSubscriptionCreationOverride) |
| && Objects.equals(persistence, other.persistence) && Objects.equals(bundles, other.bundles) |
| && Objects.equals(latency_stats_sample_rate, other.latency_stats_sample_rate) |
| && Objects.equals(message_ttl_in_seconds, |
| other.message_ttl_in_seconds) |
| && Objects.equals(subscription_expiration_time_minutes, other.subscription_expiration_time_minutes) |
| && Objects.equals(retention_policies, other.retention_policies) |
| && Objects.equals(encryption_required, other.encryption_required) |
| && Objects.equals(delayed_delivery_policies, other.delayed_delivery_policies) |
| && Objects.equals(inactive_topic_policies, other.inactive_topic_policies) |
| && Objects.equals(subscription_auth_mode, other.subscription_auth_mode) |
| && Objects.equals(max_producers_per_topic, other.max_producers_per_topic) |
| && Objects.equals(max_consumers_per_topic, other.max_consumers_per_topic) |
| && Objects.equals(max_unacked_messages_per_consumer, other.max_unacked_messages_per_consumer) |
| && Objects.equals(max_unacked_messages_per_subscription, max_unacked_messages_per_subscription) |
| && Objects.equals(max_consumers_per_subscription, max_consumers_per_subscription) |
| && Objects.equals(compaction_threshold, compaction_threshold) |
| && offload_threshold == other.offload_threshold |
| && Objects.equals(offload_deletion_lag_ms, other.offload_deletion_lag_ms) |
| && schema_auto_update_compatibility_strategy == other.schema_auto_update_compatibility_strategy |
| && schema_validation_enforced == other.schema_validation_enforced |
| && schema_compatibility_strategy == other.schema_compatibility_strategy |
| && is_allow_auto_update_schema == other.is_allow_auto_update_schema |
| && Objects.equals(offload_policies, other.offload_policies) |
| && Objects.equals(subscription_types_enabled, other.subscription_types_enabled); |
| } |
| |
| return false; |
| } |
| |
| public static BundlesData defaultBundle() { |
| BundlesData bundle = new BundlesData(1); |
| List<String> boundaries = Lists.newArrayList(); |
| boundaries.add(FIRST_BOUNDARY); |
| boundaries.add(LAST_BOUNDARY); |
| bundle.setBoundaries(boundaries); |
| return bundle; |
| } |
| |
| public static void setStorageQuota(Policies polices, BacklogQuota quota) { |
| if (polices == null) { |
| return; |
| } |
| polices.backlog_quota_map.put(BacklogQuota.BacklogQuotaType.destination_storage, quota); |
| } |
| |
| @Override |
| public String toString() { |
| return MoreObjects.toStringHelper(this).add("auth_policies", auth_policies) |
| .add("replication_clusters", replication_clusters) |
| .add("bundles", bundles) |
| .add("backlog_quota_map", backlog_quota_map) |
| .add("persistence", persistence) |
| .add("deduplicationEnabled", deduplicationEnabled) |
| .add("autoTopicCreationOverride", autoTopicCreationOverride) |
| .add("autoSubscriptionCreationOverride", autoSubscriptionCreationOverride) |
| .add("clusterDispatchRate", clusterDispatchRate) |
| .add("topicDispatchRate", topicDispatchRate) |
| .add("subscriptionDispatchRate", subscriptionDispatchRate) |
| .add("replicatorDispatchRate", replicatorDispatchRate) |
| .add("clusterSubscribeRate", clusterSubscribeRate) |
| .add("publishMaxMessageRate", publishMaxMessageRate) |
| .add("latency_stats_sample_rate", latency_stats_sample_rate) |
| .add("message_ttl_in_seconds", message_ttl_in_seconds) |
| .add("subscription_expiration_time_minutes", subscription_expiration_time_minutes) |
| .add("retention_policies", retention_policies) |
| .add("deleted", deleted) |
| .add("encryption_required", encryption_required) |
| .add("delayed_delivery_policies", delayed_delivery_policies) |
| .add("inactive_topic_policies", inactive_topic_policies) |
| .add("subscription_auth_mode", subscription_auth_mode) |
| .add("max_producers_per_topic", max_producers_per_topic) |
| .add("max_consumers_per_topic", max_consumers_per_topic) |
| .add("max_consumers_per_subscription", max_consumers_per_topic) |
| .add("max_unacked_messages_per_consumer", max_unacked_messages_per_consumer) |
| .add("max_unacked_messages_per_subscription", max_unacked_messages_per_subscription) |
| .add("compaction_threshold", compaction_threshold) |
| .add("offload_threshold", offload_threshold) |
| .add("offload_deletion_lag_ms", offload_deletion_lag_ms) |
| .add("schema_auto_update_compatibility_strategy", schema_auto_update_compatibility_strategy) |
| .add("schema_validation_enforced", schema_validation_enforced) |
| .add("schema_compatibility_Strategy", schema_compatibility_strategy) |
| .add("is_allow_auto_update_Schema", is_allow_auto_update_schema) |
| .add("offload_policies", offload_policies) |
| .add("subscription_types_enabled", subscription_types_enabled).toString(); |
| } |
| |
| private static final long MAX_BUNDLES = ((long) 1) << 32; |
| |
| public static BundlesData getBundles(int numBundles) { |
| if (numBundles <= 0) { |
| throw new RestException(Status.BAD_REQUEST, |
| "Invalid number of bundles. Number of numbles has to be in the range of (0, 2^32]."); |
| } |
| Long maxVal = MAX_BUNDLES; |
| Long segSize = maxVal / numBundles; |
| List<String> partitions = Lists.newArrayList(); |
| partitions.add(String.format("0x%08x", 0L)); |
| Long curPartition = segSize; |
| for (int i = 0; i < numBundles; i++) { |
| if (i != numBundles - 1) { |
| partitions.add(String.format("0x%08x", curPartition)); |
| } else { |
| partitions.add(String.format("0x%08x", maxVal - 1)); |
| } |
| curPartition += segSize; |
| } |
| return new BundlesData(partitions); |
| } |
| } |
