/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.ceresdb;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import io.ceresdb.proto.internal.Common;
import io.ceresdb.proto.internal.Storage;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.runners.MockitoJUnitRunner;

import io.ceresdb.common.Endpoint;
import io.ceresdb.common.util.Clock;
import io.ceresdb.models.Err;
import io.ceresdb.models.FieldValue;
import io.ceresdb.models.Result;
import io.ceresdb.models.Rows;
import io.ceresdb.models.Series;
import io.ceresdb.models.WriteOk;
import io.ceresdb.options.WriteOptions;
import io.ceresdb.rpc.Context;
import io.ceresdb.rpc.Observer;
import io.ceresdb.util.TestUtil;

/**
 * @author jiachun.fjc
 */
@RunWith(value = MockitoJUnitRunner.class)
public class WriteClientTest {

    private WriteClient  writeClient;
    @Mock
    private RouterClient routerClient;

    @Before
    public void before() {
        final WriteOptions writeOpts = new WriteOptions();
        writeOpts.setAsyncPool(ForkJoinPool.commonPool());
        writeOpts.setRoutedClient(this.routerClient);

        this.writeClient = new WriteClient();
        this.writeClient.init(writeOpts);
    }

    @After
    public void after() {
        this.writeClient.shutdownGracefully();
        this.routerClient.shutdownGracefully();
    }

    @Test
    public void writeAllSuccessTest() throws ExecutionException, InterruptedException {
        final List<Rows> data = TestUtil.newListOfRows("write_client_test_metric1", //
                "write_client_test_metric2", //
                "write_client_test_metric3");

        final Endpoint ep1 = Endpoint.of("127.0.0.1", 8081);
        final Endpoint ep2 = Endpoint.of("127.0.0.2", 8081);
        final Endpoint ep3 = Endpoint.of("127.0.0.3", 8081);

        final Common.ResponseHeader header = Common.ResponseHeader.newBuilder() //
                .setCode(Result.SUCCESS) //
                .build();
        final Storage.WriteResponse resp = Storage.WriteResponse.newBuilder() //
                .setHeader(header) //
                .setSuccess(2) //
                .build();
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep1), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep2), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep3), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.routeFor(Mockito.any())) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = 4694599978937545735L;

                    {
                        put("write_client_test_metric1", Route.of("write_client_test_metric1", ep1));
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));

        final CompletableFuture<Result<WriteOk, Err>> f = this.writeClient.write(data, Context.newDefault());
        final Result<WriteOk, Err> ret = f.get();

        Assert.assertTrue(ret.isOk());
        Assert.assertEquals(new Integer(6), ret.mapOr(0, WriteOk::getSuccess));
        Assert.assertEquals(new Integer(0), ret.mapOr(-1, WriteOk::getFailed));
    }

    @Test
    public void write3And1InvalidRoute() throws ExecutionException, InterruptedException {
        final List<Rows> data = TestUtil.newListOfRows("write_client_test_metric1", //
                "write_client_test_metric2", //
                "write_client_test_metric3");

        final Endpoint ep1 = Endpoint.of("127.0.0.1", 8081);
        final Endpoint ep2 = Endpoint.of("127.0.0.2", 8081);
        final Endpoint ep3 = Endpoint.of("127.0.0.3", 8081);
        final Endpoint ep4 = Endpoint.of("127.0.0.4", 8081);

        final Storage.WriteResponse resp = TestUtil.newSuccessWriteResp(2);
        final Storage.WriteResponse errResp = TestUtil.newFailedWriteResp(Result.INVALID_ROUTE, 2);

        Mockito.when(this.routerClient.invoke(Mockito.eq(ep1), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep2), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep3), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(errResp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep4), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.routeFor(Mockito.eq(TestUtil.asSet("write_client_test_metric1", //
                "write_client_test_metric2", "write_client_test_metric3")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -8646902388192715970L;

                    {
                        put("write_client_test_metric1", Route.of("write_client_test_metric1", ep1));
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient.routeRefreshFor(Mockito.any()))
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -3271323053870289591L;

                    {
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep4));
                    }
                }));
        Mockito.when(this.routerClient.routeFor(Mockito.eq(TestUtil.asSet("write_client_test_metric3")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = 4340010451723257789L;

                    {
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep4));
                    }
                }));

        final CompletableFuture<Result<WriteOk, Err>> f = this.writeClient.write(data, Context.newDefault());
        final Result<WriteOk, Err> ret = f.get();

        Assert.assertTrue(ret.isOk());
        Assert.assertEquals(new Integer(6), ret.mapOr(0, WriteOk::getSuccess));
        Assert.assertEquals(new Integer(0), ret.mapOr(-1, WriteOk::getFailed));
    }

    @Test
    public void write3And1InvalidRouteAndRetryFailed() throws ExecutionException, InterruptedException {
        final List<Rows> data = TestUtil.newListOfRows("write_client_test_metric1", //
                "write_client_test_metric2", //
                "write_client_test_metric3");

        final Endpoint ep1 = Endpoint.of("127.0.0.1", 8081);
        final Endpoint ep2 = Endpoint.of("127.0.0.2", 8081);
        final Endpoint ep3 = Endpoint.of("127.0.0.3", 8081);

        final Storage.WriteResponse resp = TestUtil.newSuccessWriteResp(2);
        final Storage.WriteResponse errResp = TestUtil.newFailedWriteResp(Result.INVALID_ROUTE, 2);

        Mockito.when(this.routerClient.invoke(Mockito.eq(ep1), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep2), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep3), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(errResp));
        Mockito.when(this.routerClient.routeFor(Mockito.eq(
                TestUtil.asSet("write_client_test_metric1", "write_client_test_metric2", "write_client_test_metric3")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -7535390185627686991L;

                    {
                        put("write_client_test_metric1", Route.of("write_client_test_metric1", ep1));
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient.routeRefreshFor(Mockito.any()))
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -3191375160670801662L;

                    {
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient.routeFor(Mockito.eq(TestUtil.asSet("write_client_test_metric3")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = 1341458669202248824L;

                    {
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));

        final CompletableFuture<Result<WriteOk, Err>> f = this.writeClient.write(data, Context.newDefault());
        final Result<WriteOk, Err> ret = f.get();

        final int success = ret.mapOrElse(err -> -1, WriteOk::getSuccess);
        Assert.assertEquals(-1, success);
        Assert.assertFalse(ret.isOk());
        Assert.assertEquals(4, ret.getErr().getSubOk().getSuccess());
        Assert.assertEquals(0, ret.getErr().getSubOk().getFailed());
        Assert.assertEquals(1, ret.getErr().getFailedWrites().size());
    }

    @Test
    public void write3And2FailedAndRetryFailed() throws ExecutionException, InterruptedException {
        final List<Rows> data = TestUtil.newListOfRows("write_client_test_metric1", //
                "write_client_test_metric2", //
                "write_client_test_metric3");

        final Endpoint ep1 = Endpoint.of("127.0.0.1", 8081);
        final Endpoint ep2 = Endpoint.of("127.0.0.2", 8081);
        final Endpoint ep3 = Endpoint.of("127.0.0.3", 8081);

        final Storage.WriteResponse resp = TestUtil.newSuccessWriteResp(2);
        final Storage.WriteResponse errResp = TestUtil.newFailedWriteResp(Result.SHOULD_RETRY, 2);

        Mockito.when(this.routerClient.invoke(Mockito.eq(ep1), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep2), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(errResp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep3), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(errResp));
        Mockito.when(this.routerClient.routeFor(Mockito.eq(
                TestUtil.asSet("write_client_test_metric1", "write_client_test_metric2", "write_client_test_metric3")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -5936788008084035345L;

                    {
                        put("write_client_test_metric1", Route.of("write_client_test_metric1", ep1));
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient.routeRefreshFor(Mockito.any()))
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -4748944007591733357L;

                    {
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient
                .routeFor(Mockito.eq(TestUtil.asSet("write_client_test_metric2", "write_client_test_metric3")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -1811964578845864624L;

                    {
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient
                .routeFor(Mockito.eq(TestUtil.asSet("write_client_test_metric3", "write_client_test_metric2")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = 3940955382371644111L;

                    {
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));

        final CompletableFuture<Result<WriteOk, Err>> f = this.writeClient.write(data, Context.newDefault());
        final Result<WriteOk, Err> ret = f.get();

        Assert.assertFalse(ret.isOk());
        Assert.assertEquals(2, ret.getErr().getSubOk().getSuccess());
        Assert.assertEquals(0, ret.getErr().getSubOk().getFailed());
        Assert.assertEquals(1, ret.getErr().getFailedWrites().size());
        Assert.assertTrue(ret.getErr().stream().findFirst().isPresent());
        Assert.assertEquals(1, ret.getErr().stream().findFirst().get().getFailedWrites().size());
    }

    @Test
    public void write3And2FailedAndSomeNoRetry() throws ExecutionException, InterruptedException {
        final List<Rows> data = TestUtil.newListOfRows("write_client_test_metric1", //
                "write_client_test_metric2", //
                "write_client_test_metric3");

        final Endpoint ep1 = Endpoint.of("127.0.0.1", 8081);
        final Endpoint ep2 = Endpoint.of("127.0.0.2", 8081);
        final Endpoint ep3 = Endpoint.of("127.0.0.3", 8081);

        final Storage.WriteResponse resp = TestUtil.newSuccessWriteResp(2);
        final Storage.WriteResponse errResp1 = TestUtil.newFailedWriteResp(Result.SHOULD_RETRY, 2);
        final Storage.WriteResponse errResp2 = TestUtil.newFailedWriteResp(400, 2);

        Mockito.when(this.routerClient.invoke(Mockito.eq(ep1), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep2), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(errResp1));
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep3), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(errResp2));
        Mockito.when(this.routerClient.routeFor(Mockito.eq(
                TestUtil.asSet("write_client_test_metric1", "write_client_test_metric2", "write_client_test_metric3")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = 1040769477529210661L;

                    {
                        put("write_client_test_metric1", Route.of("write_client_test_metric1", ep1));
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient.routeRefreshFor(Mockito.any()))
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -6892083230027668740L;

                    {
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep3));
                    }
                }));
        Mockito.when(this.routerClient.routeFor(Mockito.eq(TestUtil.asSet("write_client_test_metric2")))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = -9174308983134252825L;

                    {
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep2));
                    }
                }));

        final CompletableFuture<Result<WriteOk, Err>> f = this.writeClient.write(data, Context.newDefault());
        final Result<WriteOk, Err> ret = f.get();

        Assert.assertFalse(ret.isOk());
        Assert.assertEquals(2, ret.getErr().getSubOk().getSuccess());
        Assert.assertEquals(0, ret.getErr().getSubOk().getFailed());
        Assert.assertEquals(1, ret.getErr().getFailedWrites().size());
        Assert.assertTrue(ret.getErr().stream().findFirst().isPresent());
        Assert.assertEquals(1, ret.getErr().stream().findFirst().get().getFailedWrites().size());
    }

    @Test
    public void writeSplitTest() throws ExecutionException, InterruptedException {
        writeSplit(1, 2);
        writeSplit(2, 2);
        writeSplit(3, 2);
        writeSplit(4, 4);
        writeSplit(5, 4);
        writeSplit(8, 8);
    }

    private void writeSplit(final int maxWriteSize, final int partOfSuccess)
            throws ExecutionException, InterruptedException {
        // re-init
        this.writeClient.shutdownGracefully();
        final WriteOptions writeOpts = new WriteOptions();
        writeOpts.setAsyncPool(ForkJoinPool.commonPool());
        writeOpts.setRoutedClient(this.routerClient);
        writeOpts.setMaxWriteSize(maxWriteSize);
        this.writeClient = new WriteClient();
        this.writeClient.init(writeOpts);

        final List<Rows> data = TestUtil.newListOfRows("write_client_test_metric1", //
                "write_client_test_metric2", //
                "write_client_test_metric3", //
                "write_client_test_metric4", //
                "write_client_test_metric5", //
                "write_client_test_metric6", //
                "write_client_test_metric7", //
                "write_client_test_metric8");

        final Endpoint ep = Endpoint.of("127.0.0.1", 8081);

        final Common.ResponseHeader header = Common.ResponseHeader.newBuilder() //
                .setCode(Result.SUCCESS) //
                .build();
        final Storage.WriteResponse resp = Storage.WriteResponse.newBuilder() //
                .setHeader(header) //
                .setSuccess(partOfSuccess) //
                .build();
        Mockito.when(this.routerClient.invoke(Mockito.eq(ep), Mockito.any(), Mockito.any())) //
                .thenReturn(Utils.completedCf(resp));
        Mockito.when(this.routerClient.routeFor(Mockito.any())) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {
                    private static final long serialVersionUID = 4694599978937545735L;

                    {
                        put("write_client_test_metric1", Route.of("write_client_test_metric1", ep));
                        put("write_client_test_metric2", Route.of("write_client_test_metric2", ep));
                        put("write_client_test_metric3", Route.of("write_client_test_metric3", ep));
                        put("write_client_test_metric4", Route.of("write_client_test_metric4", ep));
                        put("write_client_test_metric5", Route.of("write_client_test_metric5", ep));
                        put("write_client_test_metric6", Route.of("write_client_test_metric6", ep));
                        put("write_client_test_metric7", Route.of("write_client_test_metric7", ep));
                        put("write_client_test_metric8", Route.of("write_client_test_metric8", ep));
                    }
                }));

        final CompletableFuture<Result<WriteOk, Err>> f = this.writeClient.write(data, Context.newDefault());
        final Result<WriteOk, Err> ret = f.get();

        Assert.assertTrue(ret.isOk());
        Assert.assertEquals(new Integer(16), ret.mapOr(0, WriteOk::getSuccess));
        Assert.assertEquals(new Integer(0), ret.mapOr(-1, WriteOk::getFailed));
    }

    @Test
    public void streamWriteTest() {
        final String testMetric = "stream_metric_test";
        final Endpoint ep = Endpoint.of("127.0.0.1", 8081);
        Mockito.when(this.routerClient.routeFor(Mockito.eq(Collections.singleton(testMetric)))) //
                .thenReturn(Utils.completedCf(new HashMap<String, Route>() {

                    private static final long serialVersionUID = 8473563130528272901L;

                    {
                        put(testMetric, Route.of(testMetric, ep));
                    }
                }));
        final CompletableFuture<WriteOk> f = new CompletableFuture<>();
        final AtomicInteger dataCount = new AtomicInteger();
        Mockito.when(this.routerClient.invokeClientStreaming(Mockito.any(), Mockito.any(Storage.WriteRequest.class),
                Mockito.any(), Mockito.any())).thenReturn(new Observer<Storage.WriteRequest>() {

                    @Override
                    public void onNext(final Storage.WriteRequest value) {
                        final int c = value.getMetricsList().stream() //
                                .flatMap(wmc -> wmc.getEntriesList().stream()) //
                                .map(Storage.WriteEntry::getFieldGroupsCount).reduce(0, Integer::sum);
                        dataCount.addAndGet(c);
                    }

                    @Override
                    public void onError(final Throwable err) {
                        // ignored
                    }

                    @Override
                    public void onCompleted() {
                        f.complete(WriteOk.ok(dataCount.get(), 0, null));
                    }
                });

        final StreamWriteBuf<Rows, WriteOk> writer = this.writeClient.streamWrite(testMetric);
        final CompletableFuture<WriteOk> ret = writer //
                .write(TestUtil.newRow(testMetric)) //
                .write(TestUtil.newRow(testMetric)) //
                .write(TestUtil.newRow(testMetric)) //
                .flush() //
                .write(TestUtil.newRow(testMetric)) //
                .flush() //
                .writeAndFlush(TestUtil.newListOfRows(testMetric, testMetric)) //
                .completed();

        f.whenComplete((r, e) -> {
            if (e != null) {
                ret.completeExceptionally(e);
            } else {
                ret.complete(r);
            }
        });

        Assert.assertEquals(12, ret.join().getSuccess());
    }

    @Test
    public void rowsToWriteProtoTest() {
        final Rows rs1 = Series.newBuilder("metric1") //
                .tag("m1_tag1", "v1") //
                .tag("m1_tag2", "v2") //
                .tag("m1_tag3", "v3") //
                .toRowsBuilder(true) //
                .fields(Clock.defaultClock().getTick(), input -> {
                    input.put("m1_field1", FieldValue.withDouble(0.1));
                    input.put("m1_field2", FieldValue.withString("he"));
                    input.put("m1_field3", FieldValue.withStringOrNull(null)); // null
                }) //
                .fields(Clock.defaultClock().getTick() + 10, input -> {
                    input.put("m1_field1", FieldValue.withDouble(0.1));
                    input.put("m1_field2", FieldValue.withString("he"));
                    input.put("m1_field3", FieldValue.withString("surprise!!!"));
                }) //
                .build();

        final Rows rs2 = Series.newBuilder("metric2") //
                .tag("m2_tag1", "v1") //
                .tag("m2_tag2", "v2") //
                .tag("m3_tag3", "v3") //
                .toRowsBuilder(true) //
                .fields(Clock.defaultClock().getTick(), input -> {
                    input.put("m2_field1", FieldValue.withDouble(0.1));
                    input.put("m2_field2", FieldValue.withString("he"));
                }) //
                .fields(Clock.defaultClock().getTick() + 10, input -> {
                    input.put("m2_field1", FieldValue.withDouble(0.1));
                    input.put("m2_field2", FieldValue.withString("he"));
                    input.put("m2_field3", FieldValue.withString("surprise!!!"));
                }) //
                .build();

        final Rows rs3 = Series.newBuilder("metric1") //
                .tag("m1_tag1", "vv1") //
                .tag("m1_tag2", "vv2") //
                .toRowsBuilder(true) //
                .fields(System.currentTimeMillis(), input -> {
                    input.put("m1_field1", FieldValue.withDouble(0.1));
                    input.put("m1_field2", FieldValue.withString("he"));
                }) //
                .build();

        final Storage.WriteRequest writeReq = this.writeClient.toWriteRequestObj(Stream.of(rs1, rs2, rs3));

        Assert.assertNotNull(writeReq);
        Assert.assertEquals(2, writeReq.getMetricsCount());
        final List<String> metrics = writeReq.getMetricsList() //
                .stream() //
                .map(Storage.WriteMetric::getMetric) //
                .collect(Collectors.toList());
        Assert.assertEquals(Arrays.asList("metric1", "metric2"), metrics);

        {
            final Optional<Storage.WriteMetric> opl = writeReq.getMetricsList().stream()
                    .filter(wm -> wm.getMetric().equals("metric1")) //
                    .findFirst(); //
            Assert.assertTrue(opl.isPresent());

            final Storage.WriteMetric metric = opl.get();
            Assert.assertEquals(3, metric.getTagNamesCount());
            Assert.assertEquals(3, metric.getFieldNamesCount());
            Assert.assertEquals(2, metric.getEntriesCount());

            {
                final Optional<Storage.WriteEntry> opl2 = metric.getEntriesList().stream().filter(w -> {
                    final Set<String> tagVSet = w.getTagsList().stream().map(tag -> tag.getValue().getStringValue())
                            .collect(Collectors.toSet());

                    return tagVSet.contains("v1") && tagVSet.contains("v2") && tagVSet.contains("v3");
                }).findFirst();
                Assert.assertTrue(opl2.isPresent());
                Assert.assertEquals(0, opl2.get().getTags(0).getNameIndex());
                Assert.assertEquals(1, opl2.get().getTags(1).getNameIndex());
                Assert.assertEquals(2, opl2.get().getTags(2).getNameIndex());
                Assert.assertEquals(2, opl2.get().getFieldGroupsCount());
                Assert.assertEquals(2, opl2.get().getFieldGroups(0).getFieldsCount());
                Assert.assertEquals(3, opl2.get().getFieldGroups(1).getFieldsCount());
                Assert.assertEquals(0, opl2.get().getFieldGroups(0).getFields(0).getNameIndex());
                Assert.assertEquals(1, opl2.get().getFieldGroups(0).getFields(1).getNameIndex());
                Assert.assertEquals(0, opl2.get().getFieldGroups(1).getFields(0).getNameIndex());
                Assert.assertEquals(1, opl2.get().getFieldGroups(1).getFields(1).getNameIndex());
                Assert.assertEquals(2, opl2.get().getFieldGroups(1).getFields(2).getNameIndex());
            }
        }

        {
            final Optional<Storage.WriteMetric> opl = writeReq.getMetricsList().stream()
                    .filter(wm -> wm.getMetric().equals("metric2")) //
                    .findFirst(); //
            Assert.assertTrue(opl.isPresent());

            final Storage.WriteMetric metric = opl.get();
            Assert.assertEquals(3, metric.getTagNamesCount());
            Assert.assertEquals(3, metric.getFieldNamesCount());
            Assert.assertEquals(1, metric.getEntriesCount());

            {
                final Optional<Storage.WriteEntry> opl2 = metric.getEntriesList().stream().filter(w -> {
                    final Set<String> tagVSet = w.getTagsList().stream().map(tag -> tag.getValue().getStringValue())
                            .collect(Collectors.toSet());

                    return tagVSet.contains("v1") && tagVSet.contains("v2") && tagVSet.contains("v3");
                }).findFirst();
                Assert.assertTrue(opl2.isPresent());
                Assert.assertEquals(0, opl2.get().getTags(0).getNameIndex());
                Assert.assertEquals(1, opl2.get().getTags(1).getNameIndex());
                Assert.assertEquals(2, opl2.get().getTags(2).getNameIndex());
                Assert.assertEquals(2, opl2.get().getFieldGroupsCount());
                Assert.assertEquals(2, opl2.get().getFieldGroups(0).getFieldsCount());
                Assert.assertEquals(3, opl2.get().getFieldGroups(1).getFieldsCount());
                Assert.assertEquals(0, opl2.get().getFieldGroups(0).getFields(0).getNameIndex());
                Assert.assertEquals(1, opl2.get().getFieldGroups(0).getFields(1).getNameIndex());
                Assert.assertEquals(0, opl2.get().getFieldGroups(1).getFields(0).getNameIndex());
                Assert.assertEquals(1, opl2.get().getFieldGroups(1).getFields(1).getNameIndex());
                Assert.assertEquals(2, opl2.get().getFieldGroups(1).getFields(2).getNameIndex());
            }
        }
    }
}