@Test public final void getPartitionIndexForTheTopicWithASinglePartition() { assertEquals(0, partitioner.partition(TEST_TOPIC, lp.getDeviceId().toString(), lp, 1)); lp.setDeviceId("SM_12"); assertEquals(0, partitioner.partition(TEST_TOPIC, lp.getDeviceId().toString(), lp, 1)); lp.setConsumerCategory(Category.INDUSTRIAL); assertEquals(0, partitioner.partition(TEST_TOPIC, lp.getDeviceId().toString(), lp, 1)); }
public int partition(String topic, String key, LoadProfileModel value, int numPartitions) { int lastPartitionIdx = numPartitions - 1; try { if (value.getConsumerCategory() == Category.INDUSTRIAL) { return lastPartitionIdx; } else { if (lastPartitionIdx > 0) { return (Math.abs(Utils.murmur2(key.getBytes()))) % lastPartitionIdx; } else { return lastPartitionIdx; } } } catch (NullPointerException | ClassCastException e) { throw new InvalidRecordException("Key or value is missing or isn't valid."); } }
LoadProfilePartitioner { public int partition(String topic, String key, LoadProfileModel value, int numPartitions) { int lastPartitionIdx = numPartitions - 1; try { if (value.getConsumerCategory() == Category.INDUSTRIAL) { return lastPartitionIdx; } else { if (lastPartitionIdx > 0) { return (Math.abs(Utils.murmur2(key.getBytes()))) % lastPartitionIdx; } else { return lastPartitionIdx; } } } catch (NullPointerException | ClassCastException e) { throw new InvalidRecordException("Key or value is missing or isn't valid."); } } }
LoadProfilePartitioner { public int partition(String topic, String key, LoadProfileModel value, int numPartitions) { int lastPartitionIdx = numPartitions - 1; try { if (value.getConsumerCategory() == Category.INDUSTRIAL) { return lastPartitionIdx; } else { if (lastPartitionIdx > 0) { return (Math.abs(Utils.murmur2(key.getBytes()))) % lastPartitionIdx; } else { return lastPartitionIdx; } } } catch (NullPointerException | ClassCastException e) { throw new InvalidRecordException("Key or value is missing or isn't valid."); } }  }
LoadProfilePartitioner { public int partition(String topic, String key, LoadProfileModel value, int numPartitions) { int lastPartitionIdx = numPartitions - 1; try { if (value.getConsumerCategory() == Category.INDUSTRIAL) { return lastPartitionIdx; } else { if (lastPartitionIdx > 0) { return (Math.abs(Utils.murmur2(key.getBytes()))) % lastPartitionIdx; } else { return lastPartitionIdx; } } } catch (NullPointerException | ClassCastException e) { throw new InvalidRecordException("Key or value is missing or isn't valid."); } }  int partition(String topic, String key, LoadProfileModel value, int numPartitions); }
LoadProfilePartitioner { public int partition(String topic, String key, LoadProfileModel value, int numPartitions) { int lastPartitionIdx = numPartitions - 1; try { if (value.getConsumerCategory() == Category.INDUSTRIAL) { return lastPartitionIdx; } else { if (lastPartitionIdx > 0) { return (Math.abs(Utils.murmur2(key.getBytes()))) % lastPartitionIdx; } else { return lastPartitionIdx; } } } catch (NullPointerException | ClassCastException e) { throw new InvalidRecordException("Key or value is missing or isn't valid."); } }  int partition(String topic, String key, LoadProfileModel value, int numPartitions);  }