fix: address code review comments regarding instance-level tags and duplication

Author: rahulmane-goog

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/Datastore.java

@@ -643,9 +643,7 @@ AggregationResults runAggregation(
    * @param requestTags the request tags to append to existing ones
    */
   default Datastore withRequestTags(String... requestTags) {
-    ImmutableList.Builder<String> builder = ImmutableList.builder();
-    builder.addAll(getOptions().getRequestTags()).add(requestTags);
-    return getOptions().toBuilder().setTags(builder.build()).build().getService();
+    return withRequestTags(java.util.Arrays.asList(requestTags));
   }
 
   /**

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/QueryResultsImpl.java

@@ -102,7 +102,7 @@ private void sendRequest() {
     if (explainOptions != null) {
       requestPb.setExplainOptions(explainOptions);
     }
-    if (requestOptions != null) {
+    if (requestOptions != null || !datastore.getOptions().getRequestTags().isEmpty()) {
       requestPb.setRequestOptions(
           RequestOptionsHelper.createRequestOptions(datastore.getOptions(), requestOptions));
     }

java-datastore/google-cloud-datastore/src/main/java/com/google/cloud/datastore/execution/request/AggregationQueryRequestProtoPreparer.java

@@ -75,7 +75,7 @@ public RunAggregationQueryRequest prepare(QueryConfig<AggregationQuery> queryCon
     readOptionsPb.ifPresent(aggregationQueryRequestBuilder::setReadOptions);
 
     // add request options to the request.
-    if (queryConfig.getRequestOptions() != null) {
+    if (queryConfig.getRequestOptions() != null || !datastoreOptions.getRequestTags().isEmpty()) {
       aggregationQueryRequestBuilder.setRequestOptions(
           createRequestOptions(datastoreOptions, queryConfig.getRequestOptions()));
     }

java-datastore/google-cloud-datastore/src/test/java/com/google/cloud/datastore/AbstractDatastoreTest.java

@@ -1400,6 +1400,51 @@ private Predicate<RunAggregationQueryRequest> aggregationQueryWithAlias(String a
                 .getAlias());
   }
 
+  @Test
+  public void testRunQueryWithInstanceLevelRequestTags() {
+    DatastoreOptions optionsWithTags = options.toBuilder().setTags("instance-tag").build();
+    Datastore datastoreWithTags = optionsWithTags.getService();
+
+    PartitionId partitionId =
+        PartitionId.newBuilder()
+            .setProjectId(PROJECT_ID)
+            .setDatabaseId(options.getDatabaseId())
+            .setNamespaceId(options.getNamespace())
+            .build();
+
+    com.google.datastore.v1.Query queryPb =
+        com.google.datastore.v1.Query.newBuilder()
+            .addKind(com.google.datastore.v1.KindExpression.newBuilder().setName(KIND1))
+            .build();
+
+    RunQueryRequest expectedRequest =
+        RunQueryRequest.newBuilder()
+            .setProjectId(PROJECT_ID)
+            .setPartitionId(partitionId)
+            .setQuery(queryPb)
+            .setRequestOptions(
+                com.google.datastore.v1.RequestOptions.newBuilder().addRequestTags("instance-tag").build())
+            .build();
+
+    RunQueryResponse response =
+        RunQueryResponse.newBuilder()
+            .setBatch(
+                com.google.datastore.v1.QueryResultBatch.newBuilder()
+                    .addEntityResults(EntityResult.newBuilder().setEntity(ENTITY1.toPb()))
+                    .build())
+            .build();
+
+    EasyMock.expect(rpcMock.runQuery(expectedRequest)).andReturn(response);
+    EasyMock.replay(rpcFactoryMock, rpcMock);
+
+    Query<Entity> query = Query.newEntityQueryBuilder().setKind(KIND1).build();
+    QueryResults<Entity> results = datastoreWithTags.run(query);
+    assertTrue(results.hasNext());
+    assertEquals(ENTITY1, results.next());
+
+    EasyMock.verify(rpcFactoryMock, rpcMock);
+  }
+
   @Test
   public void testRunQueryWithRequestOptions() {
     com.google.datastore.v1.RequestOptions requestOptions =