/**
 * Given a list of conditions that contain Druid valid operations and
 * a list that contains those that contain any non-supported operation,
 * it outputs a triple with three different categories:
 * 1-l) condition filters on the timestamp column,
 * 2-m) condition filters that can be pushed to Druid,
 * 3-r) condition filters that cannot be pushed to Druid.
 */
private static Triple<List<RexNode>, List<RexNode>, List<RexNode>> splitFilters(
        final RexBuilder rexBuilder, final DruidQuery input, final List<RexNode> validPreds,
        final List<RexNode> nonValidPreds, final int timestampFieldIdx) {
  final List<RexNode> timeRangeNodes = new ArrayList<>();
  final List<RexNode> pushableNodes = new ArrayList<>();
  final List<RexNode> nonPushableNodes = new ArrayList<>(nonValidPreds);
  // Number of columns with the dimensions and timestamp
  for (RexNode conj : validPreds) {
    final RelOptUtil.InputReferencedVisitor visitor = new RelOptUtil.InputReferencedVisitor();
    conj.accept(visitor);
    if (visitor.inputPosReferenced.contains(timestampFieldIdx)) {
      if (visitor.inputPosReferenced.size() != 1) {
        // Complex predicate, transformation currently not supported
        nonPushableNodes.add(conj);
      } else {
        timeRangeNodes.add(conj);
      }
    } else {
      pushableNodes.add(conj);
    }
  }
  return ImmutableTriple.of(timeRangeNodes, pushableNodes, nonPushableNodes);
}
 

public void onMatch(RelOptRuleCall call) {
  final Aggregate aggregate = call.rel(0);
  final DruidQuery query = call.rel(1);
  if (!DruidQuery.isValidSignature(query.signature() + 'a')) {
    return;
  }

  if (aggregate.indicator
          || aggregate.getGroupSets().size() != 1
          || BAD_AGG.apply(ImmutableTriple.of(aggregate, (RelNode) aggregate, query))
          || !validAggregate(aggregate, query)) {
    return;
  }
  final RelNode newAggregate = aggregate.copy(aggregate.getTraitSet(),
          ImmutableList.of(Util.last(query.rels)));
  call.transformTo(DruidQuery.extendQuery(query, newAggregate));
}
 

@Override
public Stream<Triple<UUID, UUID, ServiceLevel>> streamPlaceAndAccountAndServiceLevelByPartitionId(int partitionId) {
  	try(Context ctxt = streamPlaceAndAccountAndServiceLevelByPartitionIdTimer.time()) {
        BoundStatement bs = streamPlaceAndAccountAndServiceLevelByPartitionId.bind(partitionId);
        //bs.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
        ResultSet rs = session.execute(bs);
        return stream(rs, (row) -> {
        	ServiceLevel s = ServiceLevel.BASIC;
        	String fromDb = row.getString(PlaceEntityColumns.SERVICE_LEVEL);
        	if(StringUtils.isNotBlank(fromDb)) {
        		s = ServiceLevel.valueOf(fromDb);
        	}
           return new ImmutableTriple<>(row.getUUID(BaseEntityColumns.ID), row.getUUID(PlaceEntityColumns.ACCOUNT_ID), s);
        });
     }
}
 

public GrpcHystrixCommand(String serviceName, String methodName, Boolean isEnabledFallBack) {
  super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(serviceName))//
      .andCommandKey(HystrixCommandKey.Factory.asKey(serviceName + ":" + methodName))//
      .andCommandPropertiesDefaults(
          HystrixCommandProperties.Setter().withCircuitBreakerRequestVolumeThreshold(20)// 10秒钟内至少19此请求失败，熔断器才发挥起作用
              .withCircuitBreakerSleepWindowInMilliseconds(30000)// 熔断器中断请求30秒后会进入半打开状态,放部分流量过去重试
              .withCircuitBreakerErrorThresholdPercentage(50)// 错误率达到50开启熔断保护
              .withExecutionTimeoutEnabled(false)// 禁用这里的超时
              .withFallbackEnabled(isEnabledFallBack))//
      .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(100)
          .withAllowMaximumSizeToDivergeFromCoreSize(true).withMaximumSize(Integer.MAX_VALUE)));
  this.serviceName = serviceName;
  this.methodName = methodName;
  this.start = System.currentTimeMillis();
  this.rpcContext = new ImmutableTriple<Map<String, String>, Map<String, Object>, Set<Class>>(
      RpcContext.getContext().getAttachments(), RpcContext.getContext().get(),
      RpcContext.getContext().getHoldenGroups());
  RpcContext.removeContext();
}
 

private void init(List<Tuple> inMemoryValues) {
  // lets open the files
  for (int i = 0; i < noOfFiles; i++) {
    String fileName = folder + "/part_" + i;
    Triple<List<Tuple>, Long, Long> fileParts = FileLoader.openFilePart(fileName,
        0, openBytes, keyType, dataType, deserializer);
    openList.add(new FilePart(fileParts));
  }
  // add the in-memory values to the last
  openList.add(new FilePart(new ImmutableTriple<>(inMemoryValues, 0L, 0L)));

  // lets add to the heap the first element
  for (int i = 0; i < openList.size(); i++) {
    FilePart p = openList.get(i);
    List<Tuple> list = p.keyValues.getLeft();
    if (list.size() > p.currentIndex) {
      heap.insert(list.get(p.currentIndex), i);
      p.currentIndex++;
    }
  }
}
 

private ImmutableTriple<String, String, String> findResourcePathFromFreemarkerRouteMap(String uri)
{
    //2.查找不到，则从docRoutesMap中查找  例如
    //uri:       /root/web/page/index.ftl
    //key:       /root/web
    //value:     static/
    for (String key : freemarkerRoutesMap.keySet())
    {
        if (uri.startsWith(key + "/"))
        {
            //uri:       /root/web/page/index.ftl
            //key:       /root/web
            //value:     static/
            
            //key:       /root/web/
            String subPath = StringUtils.removeStart(uri, key + "/");
            //subPath:   page/index.ftl
            //uri解码，即可完美支持中文
            subPath = URIEncoderDecoder.decode(subPath);
            return new ImmutableTriple<String, String, String>(key + "/", freemarkerRoutesMap.get(key), subPath);
        }
    }
    return null;
}
 

void configureRegexRegions(Configuration conf) throws ConfigurationException {
  List<String> regexRegionGroupList = conf.getList(REGEX_REGION_GROUP).stream().map(o -> o.toString()).collect(Collectors.toList());
  List<String> regexRegionSearchList = conf.getList(REGEX_REGION_SEARCH).stream().map(o -> o.toString()).collect(Collectors.toList());
  List<String> regexRegionReplaceList = conf.getList(REGEX_REGION_REPLACE).stream().map(o -> o.toString()).collect(Collectors.toList());

  int groupListSize = regexRegionGroupList.size();
  int searchListSize = regexRegionSearchList.size();
  int replaceListSize = regexRegionReplaceList.size();
  if (!(groupListSize == searchListSize && searchListSize == replaceListSize)) {
    // all lists must be the same size
    throw new ConfigurationException("Error initializing class. All regexRegion lists must be the same size");
  }

  groupSearchReplaceList = new ArrayList<>(groupListSize);
  for (int index = 0; index < regexRegionGroupList.size(); index++) {
    Pattern pattern = Pattern.compile(regexRegionGroupList.get(index));
    String regex = regexRegionSearchList.get(index);
    String replacement = regexRegionReplaceList.get(index);

    ImmutableTriple<Pattern, String, String> triple = new ImmutableTriple<>(pattern, regex, replacement);
    groupSearchReplaceList.add(triple);
  }
}
 

public ImmutableTriple<Country, Country, Country> generateTeamTriple() {
	Set<Country> countries = countryService.getAvailableCountriesWithoutNoneEntry();
	List<Country> availCountries = new ArrayList<Country>(countries);
	if (CollectionUtils.isEmpty(availCountries)) {
		return null;
	}

	Country countryOne = generateRandomCountry(availCountries);
	availCountries.remove(countryOne);

	if (CollectionUtils.isEmpty(availCountries)) {
		return ImmutableTriple.of(countryOne, countryOne, countryOne);
	}

	Country countryTwo = generateRandomCountry(availCountries);
	availCountries.remove(countryTwo);

	if (CollectionUtils.isEmpty(availCountries)) {
		return ImmutableTriple.of(countryOne, countryTwo, countryTwo);
	}

	Country countryThree = generateRandomCountry(availCountries);

	return ImmutableTriple.of(countryOne, countryTwo, countryThree);
}
 

@Test
public void generateTeamTriple() {
    dataBasePopulator.createRandomMatches();

    for (int i = 0; i < 100; i++) {
        ImmutableTriple<Country, Country, Country> triple = randomValueGenerator.generateTeamTriple();
        assertThat(triple).isNotNull();
        Country countryOne = triple.getLeft();
        Country countryTwo = triple.getMiddle();
        Country countryThree = triple.getRight();
        assertThat(countryOne).isNotNull();
        assertThat(countryTwo).isNotNull();
        assertThat(countryThree).isNotNull();

        assertThat(countryOne).isNotEqualTo(countryTwo);
        assertThat(countryTwo).isNotEqualTo(countryThree);
        assertThat(countryThree).isNotEqualTo(countryOne);

        assertThat(Country.NONE).isNotEqualTo(countryOne);
        assertThat(Country.NONE).isNotEqualTo(countryTwo);
        assertThat(Country.NONE).isNotEqualTo(countryThree);
    }
}
 

@Override
@SuppressWarnings("unchecked")
public Triple<List<T>, URI, List<ClientAnnotation>> fetchPartial(final URI uri, final Class<T> typeRef) {
  final ODataPropertyRequest<ClientProperty> req =
          getClient().getRetrieveRequestFactory().getPropertyRequest(uri);
    req.setPrefer(getClient().newPreferences().includeAnnotations("*"));

  final ODataRetrieveResponse<ClientProperty> res = req.execute();

  final List<T> resItems = new ArrayList<T>();

  final ClientProperty property = res.getBody();
  if (property != null && !property.hasNullValue()) {
    for (ClientValue item : property.getCollectionValue()) {
      resItems.add((T) item.asPrimitive().toValue());
    }
  }

  return new ImmutableTriple<List<T>, URI, List<ClientAnnotation>>(
          resItems, null, Collections.<ClientAnnotation>emptyList());
}
 

/**
 * Given a list of conditions that contain Druid valid operations and
 * a list that contains those that contain any non-supported operation,
 * it outputs a triple with three different categories:
 * 1-l) condition filters on the timestamp column,
 * 2-m) condition filters that can be pushed to Druid,
 * 3-r) condition filters that cannot be pushed to Druid.
 */
private static Triple<List<RexNode>, List<RexNode>, List<RexNode>> splitFilters(
    final RexBuilder rexBuilder, final DruidQuery input, final List<RexNode> validPreds,
    final List<RexNode> nonValidPreds, final int timestampFieldIdx) {
  final List<RexNode> timeRangeNodes = new ArrayList<>();
  final List<RexNode> pushableNodes = new ArrayList<>();
  final List<RexNode> nonPushableNodes = new ArrayList<>(nonValidPreds);
  // Number of columns with the dimensions and timestamp
  for (RexNode conj : validPreds) {
    final RelOptUtil.InputReferencedVisitor visitor = new RelOptUtil.InputReferencedVisitor();
    conj.accept(visitor);
    if (visitor.inputPosReferenced.contains(timestampFieldIdx)
        && visitor.inputPosReferenced.size() == 1) {
      timeRangeNodes.add(conj);
    } else {
      pushableNodes.add(conj);
    }
  }
  return ImmutableTriple.of(timeRangeNodes, pushableNodes, nonPushableNodes);
}
 

@Test
void whenUsingTriple_thenMultipleFieldsAreReturned() {

    List<Coordinates> coordinatesList = new ArrayList<>();
    coordinatesList.add(new Coordinates(1, 1, "home"));
    coordinatesList.add(new Coordinates(2, 2, "school"));
    coordinatesList.add(new Coordinates(3, 3, "hotel"));

    Coordinates target = new Coordinates(5, 5, "gym");

    ImmutableTriple<Double, Double, Double> minAvgMax = MultipleReturnValuesUsingApacheCommonsTriple.getMinAvgMaxTriple(coordinatesList, target);

    assertEquals(2.83, scaleDouble(minAvgMax.left));   //min
    assertEquals(4.24, scaleDouble(minAvgMax.middle)); //avg
    assertEquals(5.66, scaleDouble(minAvgMax.right));  //max
}
 

/**
 * @return Left:ColumnName Middle:ColumnType Right:ColumnTypeName
 */
public static Triple<List<String>, List<Integer>, List<String>> getColumnMetaData(
        Connection conn, String tableName, String column) {
    Statement statement = null;
    ResultSet rs = null;

    Triple<List<String>, List<Integer>, List<String>> columnMetaData = new ImmutableTriple<List<String>, List<Integer>, List<String>>(
            new ArrayList<String>(), new ArrayList<Integer>(),
            new ArrayList<String>());
    try {
        statement = conn.createStatement();
        String queryColumnSql = "select " + column + " from " + tableName
                + " where 1=2";

        rs = statement.executeQuery(queryColumnSql);
        ResultSetMetaData rsMetaData = rs.getMetaData();
        for (int i = 0, len = rsMetaData.getColumnCount(); i < len; i++) {

            columnMetaData.getLeft().add(rsMetaData.getColumnName(i + 1));
            columnMetaData.getMiddle().add(rsMetaData.getColumnType(i + 1));
            columnMetaData.getRight().add(
                    rsMetaData.getColumnTypeName(i + 1));
        }
        return columnMetaData;

    } catch (SQLException e) {
        throw DataXException
                .asDataXException(DBUtilErrorCode.GET_COLUMN_INFO_FAILED,
                        String.format("获取表:%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", tableName), e);
    } finally {
        DBUtil.closeDBResources(rs, statement, null);
    }
}
 

/**
 * ==============help method=============
 */
private Triple<String, String, String> getmachineInfo(String providerAndConsumerKv, String groupService) {
    String thralUrl = consulClient.getKVValue(providerAndConsumerKv).getValue().getDecodedValue();
    GrpcURL url = GrpcURL.valueOf(thralUrl);
    String flagAndIp = StringUtils.remove(providerAndConsumerKv, groupService + "/");
    String[] serverInfos = StringUtils.split(flagAndIp, "/");
    String machineFlag = serverInfos[1];
    return new ImmutableTriple<String, String, String>(machineFlag, url.getAddress(),
        url.getParameter(Constants.HTTP_PORT_KEY));
}
 

private Triple<String, String, String> getPortHostService(String serviceId) {
    String[] args = StringUtils.split(serviceId, "-");
    String hostRpcPort = args[0];
    String service = args[1];
    String version = "1.0.0";
    if (args.length > 2) {
        version = args[2];
    }
    return new ImmutableTriple<String, String, String>(hostRpcPort, service, version);
}
 

private void bfgs() {
    System.out.println("BFGS:");
    res = 0;
    int nbatches = 100;
    int nepoch = 1000;
    int batchsize = fb_h.size() / nbatches;
    for (int epoch = 0; epoch < nepoch; epoch++) {
        res = 0;
        for (int batch = 0; batch < nbatches; batch++) {
            relation_tmp = relation_vec.clone();
            entity_tmp = entity_vec.clone();
            for (int k = 0; k < batchsize; k++) {
                int i = rand_max(fb_h.size());
                int j = rand_max(entity_num);
                double pr = 1000.0 * right_num.get(fb_r.get(i))
                        / (right_num.get(fb_r.get(i)) + left_num.get(fb_r.get(i)));
                if (method == 0)
                    pr = 500;
                if (rand_max(1000) < pr) {
                    while (ok.contains(new ImmutableTriple<>(fb_h.get(i), fb_r.get(i), j)))
                        j = rand_max(entity_num);
                    train_kb(fb_h.get(i), fb_l.get(i), fb_r.get(i), fb_h.get(i), j, fb_r.get(i));
                } else {
                    while (ok.contains(new ImmutableTriple<>(j, fb_r.get(i), fb_l.get(i))))
                        j = rand_max(entity_num);
                    train_kb(fb_h.get(i), fb_l.get(i), fb_r.get(i), j, fb_l.get(i), fb_r.get(i));
                }
                norm(relation_tmp[fb_r.get(i)]);
                norm(entity_tmp[fb_h.get(i)]);
                norm(entity_tmp[fb_l.get(i)]);
                norm(entity_tmp[j]);
            }
            relation_vec = relation_tmp;
            entity_vec = entity_tmp;
        }
        System.out.println("epoch:" + epoch + ' ' + res);
    }
}
 

private void add(int e1, int e2, int r) {
    fb_h.add(e1);
    fb_r.add(r);
    fb_l.add(e2);
    Triple<Integer, Integer, Integer> triple = new ImmutableTriple<>(e1, r, e2);
    ok.add(triple);
}
 

private void prepare() {
    List<String> entities = new ArrayList<>();
    List<String> relations = new ArrayList<>();
    List<Triple<String, String, String>> triples = new ArrayList<>();
    try (Transaction tx = db.beginTx()) {
        for (Node node : db.getAllNodes()) {
            if (!node.hasLabel(Label.label(JavaCodeExtractor.CLASS)) &&
                    !node.hasLabel(Label.label(JavaCodeExtractor.INTERFACE)) &&
                    !node.hasLabel(Label.label(JavaCodeExtractor.METHOD)) &&
                    !node.hasLabel(Label.label(JavaCodeExtractor.FIELD)))
                continue;
            entities.add("" + node.getId());
        }

        for (Relationship rel : db.getAllRelationships()) {
            Node node1 = rel.getStartNode();
            if (!node1.hasLabel(Label.label(JavaCodeExtractor.CLASS)) &&
                    !node1.hasLabel(Label.label(JavaCodeExtractor.INTERFACE)) &&
                    !node1.hasLabel(Label.label(JavaCodeExtractor.METHOD)) &&
                    !node1.hasLabel(Label.label(JavaCodeExtractor.FIELD)))
                continue;
            Node node2 = rel.getEndNode();
            if (!node2.hasLabel(Label.label(JavaCodeExtractor.CLASS)) &&
                    !node2.hasLabel(Label.label(JavaCodeExtractor.INTERFACE)) &&
                    !node2.hasLabel(Label.label(JavaCodeExtractor.METHOD)) &&
                    !node2.hasLabel(Label.label(JavaCodeExtractor.FIELD)))
                continue;
            triples.add(new ImmutableTriple<>("" + node1.getId(), "" + node2.getId(), rel.getType().name()));
            if (!relations.contains(rel.getType().name()))
                relations.add(rel.getType().name());
        }
        tx.success();
    }
    transE.prepare(entities, relations, triples);
}
 

/**
 * Registers a new document from the resultset alongside with the information whether it was a hit or not.
 * Updates the retrieval statistics.
 *
 * @param docID ID of the document that was retrieved.
 * @param k The rank of the retrieved document.
 * @param relevant Boolean that indicates whether the document was relevant (true) or not (false).
 */
public final void documentAvailable(String docID, int k, boolean relevant) {
    if (k < 1) {
      throw new IllegalArgumentException(String.format("The value k must be greater than 0 (is: %d).", k));
    }
    if (k < this.pk.size()) {
      throw new IllegalArgumentException(String.format("The provided rank %d has already been evaluated.", k));
    }
    if (relevant) {
      this.intersection += 1;
    }
    this.retrieved += 1;
    Triple<String,Float,Float> triple = new ImmutableTriple<>(docID, (float)this.intersection/(float)k,  (float)this.intersection/(float)this.relevant);
    this.pk.add(triple);
}
 

public static List<ImmutableTriple<MessageId, String, Integer>> extractIdsAndKeysAndSize(RawMessage msg)
        throws IOException {
    checkArgument(msg.getMessageIdData().getBatchIndex() == -1);

    ByteBuf payload = msg.getHeadersAndPayload();
    MessageMetadata metadata = Commands.parseMessageMetadata(payload);
    int batchSize = metadata.getNumMessagesInBatch();

    CompressionType compressionType = metadata.getCompression();
    CompressionCodec codec = CompressionCodecProvider.getCompressionCodec(compressionType);
    int uncompressedSize = metadata.getUncompressedSize();
    ByteBuf uncompressedPayload = codec.decode(payload, uncompressedSize);
    metadata.recycle();

    List<ImmutableTriple<MessageId, String, Integer>> idsAndKeysAndSize = new ArrayList<>();

    for (int i = 0; i < batchSize; i++) {
        SingleMessageMetadata.Builder singleMessageMetadataBuilder = SingleMessageMetadata.newBuilder();
        ByteBuf singleMessagePayload = Commands.deSerializeSingleMessageInBatch(uncompressedPayload,
                                                                                singleMessageMetadataBuilder,
                                                                                0, batchSize);
        MessageId id = new BatchMessageIdImpl(msg.getMessageIdData().getLedgerId(),
                                              msg.getMessageIdData().getEntryId(),
                                              msg.getMessageIdData().getPartition(),
                                              i);
        if (!singleMessageMetadataBuilder.getCompactedOut()) {
            idsAndKeysAndSize.add(ImmutableTriple.of(id, singleMessageMetadataBuilder.getPartitionKey(), singleMessageMetadataBuilder.getPayloadSize()));
        }
        singleMessageMetadataBuilder.recycle();
        singleMessagePayload.release();
    }
    uncompressedPayload.release();
    return idsAndKeysAndSize;
}
 

@Test
public void testFlowControlBatch() throws Exception {
    int numMessages = RawReaderImpl.DEFAULT_RECEIVER_QUEUE_SIZE * 5;
    String topic = "persistent://my-property/my-ns/my-raw-topic";

    publishMessages(topic, numMessages, true);

    RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
    Set<String> keys = new HashSet<>();

    while (true) {
        try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) {
            Assert.assertTrue(RawBatchConverter.isReadableBatch(m));
            List<ImmutableTriple<MessageId, String, Integer>> batchKeys = RawBatchConverter.extractIdsAndKeysAndSize(m);
            // Assert each key is unique
            for (ImmutableTriple<MessageId, String, Integer> pair : batchKeys) {
                String key = pair.middle;
                Assert.assertTrue(
                        keys.add(key),
                        "Received duplicated key '" + key + "' : already received keys = " + keys);
            }
        } catch (TimeoutException te) {
            break;
        }
    }
    Assert.assertEquals(keys.size(), numMessages);
}
 

@Test
public void testBatchingExtractKeysAndIds() throws Exception {
    String topic = "persistent://my-property/my-ns/my-raw-topic";

    try (Producer<byte[]> producer = pulsarClient.newProducer().topic(topic)
        .maxPendingMessages(3)
        .enableBatching(true)
        .batchingMaxMessages(3)
        .batchingMaxPublishDelay(1, TimeUnit.HOURS)
        .messageRoutingMode(MessageRoutingMode.SinglePartition)
        .create()) {
        producer.newMessage().key("key1").value("my-content-1".getBytes()).sendAsync();
        producer.newMessage().key("key2").value("my-content-2".getBytes()).sendAsync();
        producer.newMessage().key("key3").value("my-content-3".getBytes()).send();
    }

    RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
    try (RawMessage m = reader.readNextAsync().get()) {
        List<ImmutableTriple<MessageId, String, Integer>> idsAndKeys = RawBatchConverter.extractIdsAndKeysAndSize(m);

        Assert.assertEquals(idsAndKeys.size(), 3);

        // assert message ids are in correct order
        Assert.assertTrue(idsAndKeys.get(0).getLeft().compareTo(idsAndKeys.get(1).getLeft()) < 0);
        Assert.assertTrue(idsAndKeys.get(1).getLeft().compareTo(idsAndKeys.get(2).getLeft()) < 0);

        // assert keys are as expected
        Assert.assertEquals(idsAndKeys.get(0).getMiddle(), "key1");
        Assert.assertEquals(idsAndKeys.get(1).getMiddle(), "key2");
        Assert.assertEquals(idsAndKeys.get(2).getMiddle(), "key3");
    } finally {
        reader.closeAsync().get();
    }
}
 

@Test
public void testBatchingRebatch() throws Exception {
    String topic = "persistent://my-property/my-ns/my-raw-topic";

    try (Producer<byte[]> producer = pulsarClient.newProducer().topic(topic)
        .maxPendingMessages(3)
        .enableBatching(true)
        .batchingMaxMessages(3)
        .batchingMaxPublishDelay(1, TimeUnit.HOURS)
        .messageRoutingMode(MessageRoutingMode.SinglePartition)
        .create()) {
        producer.newMessage().key("key1").value("my-content-1".getBytes()).sendAsync();
        producer.newMessage().key("key2").value("my-content-2".getBytes()).sendAsync();
        producer.newMessage().key("key3").value("my-content-3".getBytes()).send();
    }

    RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
    try (RawMessage m1 = reader.readNextAsync().get()) {
        RawMessage m2 = RawBatchConverter.rebatchMessage(m1, (key, id) -> key.equals("key2")).get();
        List<ImmutableTriple<MessageId, String, Integer>> idsAndKeys = RawBatchConverter.extractIdsAndKeysAndSize(m2);
        Assert.assertEquals(idsAndKeys.size(), 1);
        Assert.assertEquals(idsAndKeys.get(0).getMiddle(), "key2");
        m2.close();
        Assert.assertEquals(m1.getHeadersAndPayload().refCnt(), 1);
    } finally {
        reader.closeAsync().get();
    }
}
 

public void createExtraBetForUser(String username) {
    ImmutableTriple<Country, Country, Country> triple = randomValueGenerator.generateTeamTriple();
    if (triple != null) {
        Country extraBetCountryFinalWinner = triple.getLeft();
        Country extraBetCountrySemiFinalWinner = triple.getMiddle();
        Country extraBetCountryThirdFinalWinner = triple.getRight();
        saveExtraBet(extraBetCountryFinalWinner, extraBetCountrySemiFinalWinner, extraBetCountryThirdFinalWinner,
                username);
    }
}
 

@Test
public void generateTripleButOnlyOneCountryAvailable() {
    when(countryService.getAvailableCountriesWithoutNoneEntry()).thenReturn(new HashSet<>(Collections.singletonList(Country.RUSSIA)));

    ImmutableTriple<Country, Country, Country> triple = randomValueGenerator.generateTeamTriple();
    assertThat(triple).isNotNull();
}
 

@Test
public void testRunOptimize() throws Exception {
    String outputFile = "exportkeys_test.keys";

    try {
        String splitPoint = "splitpoint";

        splitTable(splitPoint.getBytes());

        String[] argsParam = {"zookeeper", tableName, outputFile, "--optimize=1g"};
        Args args = new ManagerArgs(argsParam);
        assertEquals("zookeeper", args.getZookeeperQuorum());
        ExportKeys command = new ExportKeys(admin, args);

        waitForSplitting(2);
        command.run();

        List<Triple<String, String, String>> results = new ArrayList<>();
        for (String keys : Files.readAllLines(Paths.get(outputFile), Constant.CHARSET)) {

            String[] split = keys.split(ExportKeys.DELIMITER);
            results.add(new ImmutableTriple<>(split[0], split[1], split[2]));
        }
        assertEquals(0, results.size());
    } finally {
        Files.delete(Paths.get(outputFile));
    }
}
 

@Test
public void testRegex() throws Exception {
    String outputFile = "exportkeys_test.keys";

    try {
        String splitPoint = "splitpoint";
        splitTable(splitPoint.getBytes());
        String tableName2 = createAdditionalTable(tableName + "2");
        splitTable(tableName2, splitPoint.getBytes());

        String tableNameRegex = tableName + ".*";
        String[] argsParam = {"zookeeper", tableNameRegex, outputFile};
        Args args = new ManagerArgs(argsParam);
        assertEquals("zookeeper", args.getZookeeperQuorum());
        ExportKeys command = new ExportKeys(admin, args);

        waitForSplitting(2);
        waitForSplitting(tableName2, 2);
        command.run();

        List<Triple<String, String, String>> results = new ArrayList<>();
        for (String keys : Files.readAllLines(Paths.get(outputFile), Constant.CHARSET)) {

            String[] split = keys.split(ExportKeys.DELIMITER);
            results.add(new ImmutableTriple<>(split[0], split[1], split[2]));
        }
        assertEquals(4, results.size());
    } finally {
        Files.delete(Paths.get(outputFile));
    }
}
 

@Test
public void testRegexAll() throws Exception {
    if (miniCluster) {
        String outputFile = "exportkeys_test.keys";

        try {
            String splitPoint = "splitpoint";
            splitTable(splitPoint.getBytes());
            String tableName2 = createAdditionalTable(tableName + "2");
            splitTable(tableName2, splitPoint.getBytes());

            String[] argsParam = {"zookeeper", ".*", outputFile};
            Args args = new ManagerArgs(argsParam);
            assertEquals("zookeeper", args.getZookeeperQuorum());
            ExportKeys command = new ExportKeys(admin, args);

            waitForSplitting(2);
            waitForSplitting(tableName2, 2);
            command.run();

            List<Triple<String, String, String>> results = new ArrayList<>();
            for (String keys : Files.readAllLines(Paths.get(outputFile), Constant.CHARSET)) {

                String[] split = keys.split(ExportKeys.DELIMITER);
                results.add(new ImmutableTriple<>(split[0], split[1], split[2]));
            }
            assertEquals(4, results.size());
        } finally {
            Files.delete(Paths.get(outputFile));
        }
    }
}
 

@Test
public void testRunOptimize() throws Exception {
    String outputFile = "exportkeys_test.keys";

    try {
        String splitPoint = "splitpoint";

        splitTable(splitPoint.getBytes());

        String[] argsParam = {"zookeeper", tableName, outputFile, "--optimize=1g"};
        Args args = new ManagerArgs(argsParam);
        assertEquals("zookeeper", args.getZookeeperQuorum());
        ExportKeys command = new ExportKeys(admin, args);

        waitForSplitting(2);
        command.run();

        List<Triple<String, String, String>> results = new ArrayList<>();
        for (String keys : Files.readAllLines(Paths.get(outputFile), Constant.CHARSET)) {

            String[] split = keys.split(ExportKeys.DELIMITER);
            results.add(new ImmutableTriple<>(split[0], split[1], split[2]));
        }
        assertEquals(0, results.size());
    } finally {
        Files.delete(Paths.get(outputFile));
    }
}
 

@Test
public void testRegex() throws Exception {
    String outputFile = "exportkeys_test.keys";

    try {
        String splitPoint = "splitpoint";
        splitTable(splitPoint.getBytes());
        String tableName2 = createAdditionalTable(tableName + "2");
        splitTable(tableName2, splitPoint.getBytes());

        String tableNameRegex = tableName + ".*";
        String[] argsParam = {"zookeeper", tableNameRegex, outputFile};
        Args args = new ManagerArgs(argsParam);
        assertEquals("zookeeper", args.getZookeeperQuorum());
        ExportKeys command = new ExportKeys(admin, args);

        waitForSplitting(2);
        waitForSplitting(tableName2, 2);
        command.run();

        List<Triple<String, String, String>> results = new ArrayList<>();
        for (String keys : Files.readAllLines(Paths.get(outputFile), Constant.CHARSET)) {

            String[] split = keys.split(ExportKeys.DELIMITER);
            results.add(new ImmutableTriple<>(split[0], split[1], split[2]));
        }
        assertEquals(4, results.size());
    } finally {
        Files.delete(Paths.get(outputFile));
    }
}