public static String[] extractRecordKeysByFields(String recordKey, List<String> fields) {
    String[] fieldKV = recordKey.split(DEFAULT_RECORD_KEY_PARTS_SEPARATOR);
    return Arrays.stream(fieldKV).map(kv -> kv.split(DEFAULT_COMPOSITE_KEY_FILED_VALUE, 2))
        .filter(kvArray -> kvArray.length == 1 || fields.isEmpty() || (fields.contains(kvArray[0])))
        .map(kvArray -> {
          if (kvArray.length == 1) {
            return kvArray[0];
          } else if (kvArray[1].equals(NULL_RECORDKEY_PLACEHOLDER)) {
            return null;
          } else if (kvArray[1].equals(EMPTY_RECORDKEY_PLACEHOLDER)) {
            return "";
          } else {
            return kvArray[1];
          }
        }).toArray(String[]::new);
  }
@Test
  public void testExtractRecordKeysWithFields() {
    List<String> fields = new ArrayList<>(1);
    fields.add("id2");

    String[] s1 = KeyGenUtils.extractRecordKeysByFields("id1:1,id2:2,id3:3", fields);
    Assertions.assertArrayEquals(new String[] {"2"}, s1);

    String[] s2 = KeyGenUtils.extractRecordKeysByFields("id1:1,id2:2,2,id3:3", fields);
    Assertions.assertArrayEquals(new String[] {"2", "2"}, s2);
  }