Thanks Nico - Thanks for the feedback, and nice catch on the missing volatile.
Ron — Ron Crocker Principal Engineer & Architect ( ( •)) New Relic rcroc...@newrelic.com M: +1 630 363 8835 > On Nov 3, 2017, at 7:48 AM, Nico Kruber <n...@data-artisans.com> wrote: > > Hi Ron, > imho your code should be fine (except for a potential visibility problem on > the > changes of the non-volatile partitionMap member, depending on your needs). > > The #open() method should be called (once) for each sink initialization > (according to the javadoc) and then you should be fine with the asynchronous > updater thread. > I'm including Gordon (cc'd) just to be sure as he may know more. > > > Nico > > On Friday, 3 November 2017 04:06:02 CET Ron Crocker wrote: >> We have a system where the Kafka partition a message should go into is a >> function of a value in the message. Often, it’s value % # partitions, but >> for some values it’s not - it’s a specified list of partitions that changes >> over time. Our “simple Java library” that produces messages for this system >> also has a background thread that periodically polls a HTTP endpoint (at a >> rate of 1/minute as its default) to refresh that list of special cases. >> >> It’s easy to create a FlinkKafkaPartitioner that does the mod operation; >> what I’m not so sure about is how to get this polling operation into the >> partitioner. I’m about to try it the obvious way (create a background >> thread that polls the URL and updates the partition map), but I wonder if >> that’s actually going to cause a bunch of problems for the Flink runtime. >> >> Here’s the code that I have right now: >> public class EventInsertPartitioner extends KafkaPartitioner<Tuple2<Long, >> String>> { private final String partitionerURL; >> private final long updateIntervalInMillis; >> private Map<Long, List<Integer>> partitionMap; >> private ScheduledExecutorService executor; >> >> public EventInsertPartitioner(String partitionerURL, long >> updateIntervalInMillis) { this.partitionerURL = partitionerURL; >> this.updateIntervalInMillis = updateIntervalInMillis; >> this.partitionMap = new HashMap<>(); >> } >> >> @Override >> public void open(int parallelInstanceId, int parallelInstances, int[] >> partitions) { executor = Executors.newScheduledThreadPool(1); >> executor.scheduleAtFixedRate( >> () -> updatePartitionMapRunnable(), >> updateIntervalInMillis, >> updateIntervalInMillis, >> TimeUnit.MILLISECONDS); >> >> } >> >> private void updatePartitionMapRunnable() { >> // Make synchronous request to partitionerURL >> // This is a simple JSON that matches our data >> String response = "{1:[1,2,3],2:[2]}"; >> // Replace current partitionMap with new HashMap from the response >> this.partitionMap = convertResponseToMap(response); >> // Replacing the current value of partitionMap with the updated >> version doesn't // require synchronization >> } >> >> private Map<Long, List<Integer>> convertResponseToMap(String response) { >> Map<Long, List<Integer>> hashMap = new HashMap<>(); >> // Convert response to JSON structure and just use that? >> // or Iterate and add to local hashMap >> return hashMap; >> } >> >> @Override >> public int partition(Tuple2<Long, String> next, byte[] serializedKey, >> byte[] serializedValue, int numPartitions) { long myKey = next.f0; >> >> if (partitionMap.containsKey(myKey)) { >> List<Integer> partitions = partitionMap.get(myKey); >> myKey = >> partitions.get(ThreadLocalRandom.current().nextInt(partitions.size())); } >> >> return (int)(myKey % numPartitions); >> } >> } >> Ron >> — >> Ron Crocker >> Principal Engineer & Architect >> ( ( •)) New Relic >> rcroc...@newrelic.com >> M: +1 630 363 8835