Add Weighted Id Generator #34
Conversation
| protected final Function<String, Integer> partitionResolver; | ||
| protected final int partitionCount; | ||
|
|
||
| // dataStore Structure |
There was a problem hiding this comment.
Commented code, can be removed
| */ | ||
| @SuppressWarnings("unused") | ||
| @Slf4j | ||
| public class DistributedIdGenerator { |
There was a problem hiding this comment.
Maybe rename it to partitionAwareIdGenerator.
| private static final Map<String, List<KeyValidationConstraint>> DOMAIN_SPECIFIC_CONSTRAINTS = new HashMap<>(); | ||
| protected static final int NODE_ID = IdGenerator.getNodeId(); | ||
| @Getter | ||
| private final Map<String, Map<Long, PartitionIdTracker>> dataStore = new ConcurrentHashMap<>(); |
There was a problem hiding this comment.
Maybe rename it to IdPartitionsMapping or IdPartitionStore, datastore doesn't capture what this is storing.
| = skipGlobal | ||
| ? null | ||
| : GLOBAL_CONSTRAINTS.stream() | ||
| .filter(constraint -> !constraint.isValid(partitionId)) |
There was a problem hiding this comment.
Here we are assuming that constraints will always be on partitionId, Should we have it on ID level also, else client have to implement this method, and can't moveover seamlessly. Let's discuss it once.
| val idPool = partitionIdTracker.getPartition(targetPartitionId); | ||
| int idIdx = idPool.getPointer().getAndIncrement(); | ||
| // ToDo: Add Retry Limit | ||
| while (idPool.getIds().size() <= idIdx) { |
There was a problem hiding this comment.
Let's add a retry limit which is explicitly passed by client, since partitionResolver is passed by client.
| } | ||
|
|
||
| private Optional<Integer> getTargetPartitionId(final List<KeyValidationConstraint> inConstraints, final boolean skipGlobal) { | ||
| // ToDo: Check if we need Collision Checker here |
There was a problem hiding this comment.
Shouldn't we required, since we are fetching from list.
| private void deleteExpiredKeys() { | ||
| val timeThreshold = DateTime.now().getMillis() / 1000; | ||
| for (val entry : dataStore.entrySet()) { | ||
| entry.getValue().entrySet().removeIf(partitionIdTrackerEntry -> partitionIdTrackerEntry.getKey() < timeThreshold); |
There was a problem hiding this comment.
Need to think and handle cases where time changes on a machine, probably safer to remove entries which are timeThreshold - SOME_VALUE
| } | ||
| } | ||
|
|
||
| private void deleteExpiredKeys() { |
There was a problem hiding this comment.
Should make this method synchronised.
|
|
||
| protected int readRetryCount() { | ||
| try { | ||
| val count = Integer.parseInt(System.getenv().getOrDefault("NUM_ID_GENERATION_RETRIES", "512")); |
There was a problem hiding this comment.
Should we keep default as the number of partition?
| import org.joda.time.format.DateTimeFormat; | ||
| import org.joda.time.format.DateTimeFormatter; | ||
|
|
||
| public class DistributedIdFormatter implements IdFormatter { |
| package io.appform.ranger.discovery.bundle.id.constraints; | ||
|
|
||
|
|
||
| public interface KeyValidationConstraint { |
There was a problem hiding this comment.
maybe rename it to partitionValidationConstraint?
CyberDrudge
force-pushed
the
weighted-id-generator
branch
from
72bb652 to
5f632ec
Compare
No description provided.