Restore changes

Signed-off-by: Gao Hongtao <hanahmily@gmail.com>

Author: hanahmily

src/main/java/org/apache/skywalking/banyandb/v1/client/BanyanDBClient.java

@@ -400,30 +400,15 @@ public StreamWrite createStreamWrite(String group, String name, final String ele
     }
 
     /**
-     * Build a StreamWrite request.
+     * Build a trace bulk write processor.
      *
-     * @param group     the group of the stream
-     * @param name      the name of the stream
-     * @param elementId the primary key of the stream
-     * @param timestamp the timestamp of the stream
-     * @return the request to be built
+     * @param maxBulkSize   the max size of each flush. The actual size is determined by the length of byte array.
+     * @param flushInterval if given maxBulkSize is not reached in this period, the flush would be trigger
+     *                      automatically. Unit is second.
+     * @param concurrency   the number of concurrency would run for the flush max.
+     * @param timeout       network timeout threshold in seconds.
+     * @return trace bulk write processor
      */
-    public StreamWrite createStreamWrite(String group, String name, final String elementId, long timestamp) throws BanyanDBException {
-        Preconditions.checkArgument(!Strings.isNullOrEmpty(group));
-        Preconditions.checkArgument(!Strings.isNullOrEmpty(name));
-        return new StreamWrite(this.metadataCache.findStreamMetadata(group, name), elementId, timestamp);
-    }
-
-        /**
-         * Build a trace bulk write processor.
-         *
-         * @param maxBulkSize   the max size of each flush. The actual size is determined by the length of byte array.
-         * @param flushInterval if given maxBulkSize is not reached in this period, the flush would be trigger
-         *                      automatically. Unit is second.
-         * @param concurrency   the number of concurrency would run for the flush max.
-         * @param timeout       network timeout threshold in seconds.
-         * @return trace bulk write processor
-         */
     public TraceBulkWriteProcessor buildTraceWriteProcessor(int maxBulkSize, int flushInterval, int concurrency, int timeout) {
         checkState(this.traceServiceStub != null, "trace service is null");
 

src/main/java/org/apache/skywalking/banyandb/v1/client/StreamWrite.java

@@ -50,16 +50,6 @@ public class StreamWrite extends AbstractWrite<BanyandbStream.WriteRequest> {
         this.elementId = elementId;
     }
 
-    /**
-     * Create a StreamWrite with initial timestamp.
-     *
-     * @param timestamp in milliseconds
-     */
-    public StreamWrite(MetadataCache.EntityMetadata streamMetadata, String elementId, long timestamp) {
-        this(streamMetadata, elementId);
-        this.timestamp = Optional.of(timestamp);
-    }
-
     @Override
     public StreamWrite tag(String tagName, Serializable<BanyandbModel.TagValue> tagValue) throws BanyanDBException {
         return (StreamWrite) super.tag(tagName, tagValue);

src/test/java/org/apache/skywalking/banyandb/v1/client/ITBanyanDBStreamQueryTests.java

@@ -94,7 +94,7 @@ public void testStreamQuery_TraceID() throws BanyanDBException, ExecutionExcepti
         String dbType = "SQL";
         String dbInstance = "127.0.0.1:3306";
 
-        StreamWrite streamWrite = client.createStreamWrite("sw_record", "trace", segmentId, now.toEpochMilli())
+        StreamWrite streamWrite = client.createStreamWrite("sw_record", "trace", segmentId)
                 .tag("data_binary", Value.binaryTagValue(byteData))
                 .tag("trace_id", Value.stringTagValue(traceId)) // 0
                 .tag("state", Value.longTagValue(state)) // 1
@@ -109,6 +109,7 @@ public void testStreamQuery_TraceID() throws BanyanDBException, ExecutionExcepti
                 .tag("mq.broker", Value.stringTagValue(broker)) // 10
                 .tag("mq.topic", Value.stringTagValue(topic)) // 11
                 .tag("mq.queue", Value.stringTagValue(queue)); // 12
+        streamWrite.setTimestamp(now.toEpochMilli());
 
         CompletableFuture<Void> f = processor.add(streamWrite);
         f.exceptionally(exp -> {