将FlatMessage按指定的字段值hash拆分

@param flatMessage flatMessage
@param partitionsNum 分区数量
@param pkHashConfigs hash映射
@return 拆分后的flatMessage数组
770688
public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, String pkHashConfigs) {
        if (partitionsNum == null) {
            partitionsNum = 1;
        }
        FlatMessage[] partitionMessages = new FlatMessage[partitionsNum];

        if (flatMessage.getIsDdl()) {
            partitionMessages[0] = flatMessage;
        } else {
            if (flatMessage.getData() != null && !flatMessage.getData().isEmpty()) {
                String database = flatMessage.getDatabase();
                String table = flatMessage.getTable();
                HashMode hashMode = getPartitionHashColumns(database + "." + table, pkHashConfigs);
                if (hashMode == null) {
                    // 如果都没有匹配，发送到第一个分区
                    partitionMessages[0] = flatMessage;
                } else if (hashMode.tableHash) {
                    int hashCode = table.hashCode();
                    int pkHash = Math.abs(hashCode) % partitionsNum;
                    // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序
                    pkHash = Math.abs(pkHash);
                    partitionMessages[pkHash] = flatMessage;
                } else {
                    List<String> pkNames = hashMode.pkNames;
                    if (hashMode.autoPkHash) {
                        pkNames = flatMessage.getPkNames();
                    }

                    int idx = 0;
                    for (Map<String, String> row : flatMessage.getData()) {
                        int hashCode = database.hashCode();
                        for (String pkName : pkNames) {
                            String value = row.get(pkName);
                            if (value == null) {
                                value = "";
                            }
                            hashCode = hashCode ^ value.hashCode();
                        }

                        int pkHash = Math.abs(hashCode) % partitionsNum;
                        // math.abs可能返回负值，这里再取反，把出现负值的数据还是写到固定的分区，仍然可以保证消费顺序
                        pkHash = Math.abs(pkHash);

                        FlatMessage flatMessageTmp = partitionMessages[pkHash];
                        if (flatMessageTmp == null) {
                            flatMessageTmp = new FlatMessage(flatMessage.getId());
                            partitionMessages[pkHash] = flatMessageTmp;
                            flatMessageTmp.setDatabase(flatMessage.getDatabase());
                            flatMessageTmp.setTable(flatMessage.getTable());
                            flatMessageTmp.setIsDdl(flatMessage.getIsDdl());
                            flatMessageTmp.setType(flatMessage.getType());
                            flatMessageTmp.setSql(flatMessage.getSql());
                            flatMessageTmp.setSqlType(flatMessage.getSqlType());
                            flatMessageTmp.setMysqlType(flatMessage.getMysqlType());
                            flatMessageTmp.setEs(flatMessage.getEs());
                            flatMessageTmp.setTs(flatMessage.getTs());
                        }
                        List<Map<String, String>> data = flatMessageTmp.getData();
                        if (data == null) {
                            data = new ArrayList<>();
                            flatMessageTmp.setData(data);
                        }
                        data.add(row);
                        if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) {
                            List<Map<String, String>> old = flatMessageTmp.getOld();
                            if (old == null) {
                                old = new ArrayList<>();
                                flatMessageTmp.setOld(old);
                            }
                            old.add(flatMessage.getOld().get(idx));
                        }
                        idx++;
                    }
                }
            } else {
                // 针对stmt/mixed binlog格式的query事件
                partitionMessages[0] = flatMessage;
            }
        }
        return partitionMessages;
    }
861816
private static List<Integer> repartitionStreamlet(int incomingInteger, int numPartitions) {
    List<Integer> partitions;

    if (incomingInteger >= 0 && incomingInteger < 25) {
      partitions = Arrays.asList(0, 1);
    } else if (incomingInteger > 26 && incomingInteger < 50) {
      partitions = Arrays.asList(2, 3);
    } else if (incomingInteger > 50 && incomingInteger < 75) {
      partitions = Arrays.asList(4, 5);
    } else if (incomingInteger > 76 && incomingInteger <= 100) {
      partitions = Arrays.asList(6, 7);
    } else {
      partitions = Arrays.asList(ThreadLocalRandom.current().nextInt(0, 8));
    }

    String logMessage = String.format("Sending value %d to partitions: %s",
        incomingInteger,
        StreamletUtils.intListAsString(partitions));

    LOG.info(logMessage);

    return partitions;
  }
62835
public static int hashBytes(MemorySegment segment, int offset, int lengthInBytes) {
		return hashBytes(segment, offset, lengthInBytes, DEFAULT_SEED);
	}
152137
public void setPartitionValues(java.util.Collection<String> partitionValues) {
        if (partitionValues == null) {
            this.partitionValues = null;
            return;
        }

        this.partitionValues = new java.util.ArrayList<String>(partitionValues);
    }
1067035
private static List<NodeInfo> buildPartitionHosts(List<NodeInfo> nodeInfos, CouchbasePartitionInfo partitionInfo) {
        List<NodeInfo> partitionHosts = new ArrayList<NodeInfo>();
        for (String rawHost : partitionInfo.partitionHosts()) {
            NetworkAddress convertedHost;
            int directPort;
            try {
                String parts[] = rawHost.split(":");
                String host = "";
                String port = parts[parts.length - 1];
                if (parts.length > 2) {
                    // Handle IPv6 syntax
                    for (int i = 0; i < parts.length - 1; i++) {
                        host += parts[i];
                        if (parts[i].endsWith("]")) {
                            break;
                        } else {
                            host += ":";
                        }
                    }
                } else {
                    // Simple IPv4 Handling
                    host = parts[0];
                }

                convertedHost = NetworkAddress.create(host);
                try {
                    directPort = Integer.parseInt(port);
                } catch (NumberFormatException e) {
                    LOGGER.warn("Could not parse port from the node address: {}, fallback to 0", system(rawHost));
                    directPort = 0;
                }
            } catch (Exception e) {
                throw new ConfigurationException("Could not resolve " + rawHost + "on config building.", e);
            }
            for (NodeInfo nodeInfo : nodeInfos) {
                // Make sure we only take into account nodes which contain KV
                if (!nodeInfo.services().containsKey(ServiceType.BINARY)) {
                    continue;
                }

                if (nodeInfo.hostname().equals(convertedHost) &&
                        (nodeInfo.services().get(ServiceType.BINARY) == directPort || directPort == 0)) {
                    partitionHosts.add(nodeInfo);
                }
            }
        }
        if (partitionHosts.size() != partitionInfo.partitionHosts().length) {
            throw new ConfigurationException("Partition size is not equal after conversion, this is a bug.");
        }
        return partitionHosts;
    }
1289573
public static int hash(MemorySegment[] segments, int offset, int numBytes) {
		if (inFirstSegment(segments, offset, numBytes)) {
			return MurmurHashUtil.hashBytes(segments[0], offset, numBytes);
		} else {
			return hashMultiSeg(segments, offset, numBytes);
		}
	}
231053
protected int marshalMessagesToSameDestination(Address dest, Message[] buf,
                                                   int start_index, final int end_index, int max_bundle_size) throws Exception {
        int num_msgs=0, bytes=0;
        for(;;) {
            Message msg=buf[start_index];
            if(msg != null && Objects.equals(dest, msg.dest())) {
                long size=msg.size();
                if(bytes + size > max_bundle_size)
                    break;
                bytes+=size;
                num_msgs++;
                buf[start_index]=null;
                msg.writeToNoAddrs(msg.src(), output, transport.getId());
            }
            if(start_index == end_index)
                break;
            start_index=advance(start_index);
        }
        return num_msgs;
    }
405261
public static AbstractMessage duplicate( Message srcMessage ) throws JMSException
    {
    	AbstractMessage msgCopy;
    	
        // Internal type copy
        if (srcMessage instanceof AbstractMessage)
        	msgCopy = ((AbstractMessage)srcMessage).copy();
        else
        if (srcMessage instanceof TextMessage)
        	msgCopy = duplicateTextMessage((TextMessage)srcMessage);
        else
        if (srcMessage instanceof ObjectMessage)
        	msgCopy = duplicateObjectMessage((ObjectMessage)srcMessage);
        else
        if (srcMessage instanceof BytesMessage)
        	msgCopy = duplicateBytesMessage((BytesMessage)srcMessage);
        else
        if (srcMessage instanceof MapMessage)
        	msgCopy = duplicateMapMessage((MapMessage)srcMessage);
        else
        if (srcMessage instanceof StreamMessage)
        	msgCopy = duplicateStreamMessage((StreamMessage)srcMessage);
        else
        	msgCopy = duplicateMessage(srcMessage);

        return msgCopy;
    }
803004
public static long murmurhash3(CharSequence chars, int seed) {

		final int len = chars.length();

		final int nblocks = len / 8;

		long h1 = seed;
		long h2 = seed;

		final long c1 = 0x87c37b91114253d5L;
		final long c2 = 0x4cf5ad432745937fL;

		for (int i = 0; i < nblocks; ++i) {
			int i0 = (i*2 + 0) * 4;
			int i1 = (i*2 + 1) * 4;

			long k1 = (long)chars.charAt(i0)
				     | (long)chars.charAt(i0+1) << 16
				     | (long)chars.charAt(i0+2) << 32
				     | (long)chars.charAt(i0+3) << 48;
			long k2 = (long)chars.charAt(i1)
				     | (long)chars.charAt(i1+1) << 16
				     | (long)chars.charAt(i1+2) << 32
				     | (long)chars.charAt(i1+3) << 48;

			k1 *= c1; k1 = k1 << 31 | k1 >>> (64-31); k1 *= c2; h1 ^= k1;

			h1 = h1 << 27 | h1 >>> (64-27); h1 += h2; h1 = h1*5 + 0x52dce729;

			k2 *= c2; k2 = k2 << 33 | k2 >>> (64-33); k2 *= c1; h2 ^= k2;

			h2 = h2 << 31 | h1 >>> (64-31); h2 += h1; h2 = h2*5 + 0x38495ab5;
		}

		long k1 = 0;
		long k2 = 0;

		switch (len & 7) {
			case  7: k2 ^= (long)chars.charAt(6) << 32;
			case  6: k2 ^= (long)chars.charAt(5) << 16;
			case  5: k2 ^= (long)chars.charAt(4) << 0;
						k2 *= c2; k2 = k2 << 33 | k2 >>> (64-33); k2 *= c1; h2 ^= k2;

			case  4: k1 ^= (long)chars.charAt(3) << 48;
			case  3: k1 ^= (long)chars.charAt(2) << 32;
			case  2: k1 ^= (long)chars.charAt(1) << 16;
			case  1: k1 ^= (long)chars.charAt(0) << 0;
						k1 *= c1; k1 = k1 << 31 | k1 >>> (64-31); k1 *= c2; h1 ^= k1;
			case  0: break;
		}

		h1 ^= len; h2 ^= len;

		h1 += h2;
		h2 += h1;

		h1 = fmix(h1);
		h2 = fmix(h2);

		h1 += h2;
		// h2 += h1;

		return h1;
	}
63591
public void setPartitionKeys(List<FieldSchema> partitionKeys) {
    this.setProp(HIVE_PARTITION_KEYS, GSON.toJson(partitionKeys, FIELD_SCHEMA_TYPE));
  }
606677
def unpack_messages(self, partitions_msgs):
        

        for pmsg in partitions_msgs:
            key = pmsg['_key']
            partition = pmsg['partition']
            offset = pmsg['offset']
            msg = pmsg.pop('message')
            if msg:
                try:
                    record = self.deserialize_fun(msg)
                except Exception as e:
                    log.error('Error unpacking record at partition:offset {}:{} (key: {} : {})'.format(partition, offset, key, repr(e)))
                    continue
                else:
                    if isinstance(record, dict):
                        pmsg['record'] = record
                        yield pmsg
                    else:
                        log.info('Record {} has wrong type'.format(key))
            else:
                yield pmsg
1091547
public static void validateClusterPartitionCounts(final Cluster lhs, final Cluster rhs) {
        if(lhs.getNumberOfPartitions() != rhs.getNumberOfPartitions())
            throw new VoldemortException("Total number of partitions should be equal [ lhs cluster ("
                                         + lhs.getNumberOfPartitions()
                                         + ") not equal to rhs cluster ("
                                         + rhs.getNumberOfPartitions() + ") ]");
    }
1058513
public static ZabMessage.ClusterConfiguration
  buildConfig(ClusterConfiguration config) {
    ZabMessage.Zxid version = toProtoZxid(config.getVersion());
    return ZabMessage.ClusterConfiguration.newBuilder()
                                          .setVersion(version)
                                          .addAllServers(config.getPeers())
                                          .build();
  }
1176962
public static Message buildAckEpoch(long epoch, Zxid lastZxid) {
    ZabMessage.Zxid zxid = toProtoZxid(lastZxid);

    AckEpoch ackEpoch = AckEpoch.newBuilder()
                        .setAcknowledgedEpoch(epoch)
                        .setLastZxid(zxid)
                        .build();

    return Message.newBuilder().setType(MessageType.ACK_EPOCH)
                               .setAckEpoch(ackEpoch)
                               .build();
  }
1289414
def decompress_messages(self, partitions_offmsgs):
        

        for pomsg in partitions_offmsgs:
            if pomsg['message']:
                pomsg['message'] = self.decompress_fun(pomsg['message'])
            yield pomsg
615539
public static List<PartitionDefinition> assignPartitions(List<PartitionDefinition> partitions, int currentTask, int totalTasks) {
        int esPartitions = partitions.size();
        if (totalTasks >= esPartitions) {
            return (currentTask >= esPartitions ? Collections.<PartitionDefinition>emptyList() : Collections.singletonList(partitions.get(currentTask)));
        } else {
            int partitionsPerTask = esPartitions / totalTasks;
            int remainder = esPartitions % totalTasks;

            int partitionsPerCurrentTask = partitionsPerTask;

            // spread the reminder against the tasks
            if (currentTask < remainder) {
                partitionsPerCurrentTask++;
            }

            // find the offset inside the collection
            int offset = partitionsPerTask * currentTask;
            if (currentTask != 0) {
                offset += (remainder > currentTask ? 1 : remainder);
            }

            // common case
            if (partitionsPerCurrentTask == 1) {
                return Collections.singletonList(partitions.get(offset));
            }

            List<PartitionDefinition> pa = new ArrayList<PartitionDefinition>(partitionsPerCurrentTask);
            for (int index = offset; index < offset + partitionsPerCurrentTask; index++) {
                pa.add(partitions.get(index));
            }
            return pa;
        }
    }
815067
public int getPartition(K key, V value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }