ODP-925|Druid PR-14767 | migrating guave to v27.0-jre

acceldata-io · Dec 15, 2023 · 66b0a20 · 66b0a20
1 parent 4f998db
commit 66b0a20
Show file tree

Hide file tree

Showing 41 changed files with 167 additions and 94 deletions.
diff --git a/codestyle/guava16-forbidden-apis.txt b/codestyle/guava16-forbidden-apis.txt
@@ -1,3 +0,0 @@
-# Those signatures are only available in Guava 16:
-com.google.common.util.concurrent.MoreExecutors#sameThreadExecutor() @ Use org.apache.druid.java.util.common.concurrent.Execs#directExecutor()
-com.google.common.base.Objects#firstNonNull(java.lang.Object, java.lang.Object) @ Use org.apache.druid.common.guava.GuavaUtils#firstNonNull(java.lang.Object, java.lang.Object) instead (probably... the GuavaUtils method return object is nullable)

diff --git a/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java b/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java
@@ -1199,7 +1199,7 @@ public void onFailure(Throwable t)
             settableFuture.setException(t);
           }
         }
-      });
+      }, MoreExecutors.directExecutor());
 
       taskFutures.add(settableFuture);
     }

diff --git a/...ti-stage-query/src/main/java/org/apache/druid/msq/exec/ExceptionWrappingWorkerClient.java b/...ti-stage-query/src/main/java/org/apache/druid/msq/exec/ExceptionWrappingWorkerClient.java
@@ -23,6 +23,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import org.apache.druid.frame.channel.ReadableByteChunksFrameChannel;
 import org.apache.druid.frame.key.ClusterByPartitions;
@@ -158,7 +159,8 @@ public void onFailure(Throwable t)
           {
             retVal.setException(new MSQException(t, new WorkerRpcFailedFault(workerTaskId)));
           }
-        }
+        },
+        MoreExecutors.directExecutor()
     );
 
     return retVal;

diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java
@@ -32,6 +32,7 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import it.unimi.dsi.fastutil.bytes.ByteArrays;
 import org.apache.druid.common.guava.FutureUtils;
@@ -1309,7 +1310,8 @@ public void onFailure(final Throwable t)
                       kernelHolder.getStageKernelMap().get(stageDef.getId()).fail(t)
               );
             }
-          }
+          },
+          MoreExecutors.directExecutor()
       );
     }
 
@@ -1601,7 +1603,7 @@ public OutputChannel openNilChannel(int expectedZero)
               };
 
               // Chain futures so we only sort one partition at a time.
-              nextFuture = Futures.transform(
+              nextFuture = Futures.transformAsync(
                   nextFuture,
                   (AsyncFunction<OutputChannel, OutputChannel>) ignored -> {
                     final SuperSorter sorter = new SuperSorter(
@@ -1628,7 +1630,8 @@ public OutputChannel openNilChannel(int expectedZero)
                     );
 
                     return FutureUtils.transform(sorter.run(), r -> Iterables.getOnlyElement(r.getAllChannels()));
-                  }
+                  },
+	          MoreExecutors.directExecutor()
               );
 
               sortedChannelFutures.add(nextFuture);
@@ -1654,16 +1657,18 @@ public ListenableFuture<OutputChannels> build()
         throw new ISE("Not initialized");
       }
 
-      return Futures.transform(
+      return Futures.transformAsync(
           pipelineFuture,
           (AsyncFunction<ResultAndChannels<?>, OutputChannels>) resultAndChannels ->
               Futures.transform(
                   resultAndChannels.getResultFuture(),
                   (Function<Object, OutputChannels>) input -> {
                     sanityCheckOutputChannels(resultAndChannels.getOutputChannels());
                     return resultAndChannels.getOutputChannels();
-                  }
-              )
+		 },
+                  MoreExecutors.directExecutor()
+              ),
+          	MoreExecutors.directExecutor()
       );
     }
 
@@ -1731,7 +1736,8 @@ public void onFailure(Throwable t)
                   }
               );
             }
-          }
+          },
+          MoreExecutors.directExecutor()
       );
 
       return new ResultAndChannels<>(
@@ -1761,7 +1767,7 @@ private void pushAsync(final ExceptionalFunction<ResultAndChannels<?>, Listenabl
       }
 
       pipelineFuture = FutureUtils.transform(
-          Futures.transform(
+          Futures.transformAsync(
               pipelineFuture,
               new AsyncFunction<ResultAndChannels<?>, ResultAndChannels<?>>()
               {
@@ -1770,7 +1776,8 @@ public ListenableFuture<ResultAndChannels<?>> apply(ResultAndChannels<?> t) thro
                 {
                   return fn.apply(t);
                 }
-              }
+              },
+              MoreExecutors.directExecutor()
           ),
           resultAndChannels -> new ResultAndChannels<>(
               resultAndChannels.getResultFuture(),

diff --git a/...s-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerSketchFetcher.java b/...s-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerSketchFetcher.java
@@ -23,6 +23,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import org.apache.druid.common.guava.FutureUtils;
 import org.apache.druid.java.util.common.ISE;
@@ -210,7 +211,7 @@ public void onFailure(Throwable t)
         }
 
       }
-    });
+    }, MoreExecutors.directExecutor());
 
     FutureUtils.getUnchecked(kernelActionFuture, true);
   }

diff --git a/...i-stage-query/src/main/java/org/apache/druid/msq/indexing/client/IndexerWorkerClient.java b/...i-stage-query/src/main/java/org/apache/druid/msq/indexing/client/IndexerWorkerClient.java
@@ -24,6 +24,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import org.apache.druid.common.guava.FutureUtils;
@@ -268,7 +269,8 @@ public void onFailure(Throwable t)
           {
             retVal.setException(t);
           }
-        }
+        },
+        MoreExecutors.directExecutor()
     );
 
     return retVal;

diff --git a/...age-query/src/main/java/org/apache/druid/msq/shuffle/input/WorkerInputChannelFactory.java b/...age-query/src/main/java/org/apache/druid/msq/shuffle/input/WorkerInputChannelFactory.java
@@ -23,6 +23,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.druid.frame.channel.ReadableByteChunksFrameChannel;
 import org.apache.druid.frame.channel.ReadableFrameChannel;
 import org.apache.druid.java.util.common.StringUtils;
@@ -90,7 +91,8 @@ public void onFailure(Throwable t)
           {
             channel.setError(t);
           }
-        }
+        },
+        MoreExecutors.directExecutor()
     );
   }
 

diff --git a/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MSQFutureUtils.java b/...nsions-core/multi-stage-query/src/main/java/org/apache/druid/msq/util/MSQFutureUtils.java
@@ -22,6 +22,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.druid.common.guava.FutureUtils;
 
 import javax.annotation.Nullable;
@@ -61,7 +62,8 @@ public void onFailure(Throwable t)
                 inputFuture.cancel(true);
               }
             }
-          }
+          },
+          MoreExecutors.directExecutor()
       );
     }
 

diff --git a/.../main/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTask.java b/.../main/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTask.java
@@ -31,6 +31,7 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import org.apache.druid.data.input.Committer;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.druid.data.input.Firehose;
 import org.apache.druid.data.input.FirehoseFactory;
 import org.apache.druid.data.input.InputRow;
@@ -698,9 +699,10 @@ private void publishSegments(
         committerSupplier.get(),
         Collections.singletonList(sequenceName)
     );
-    pendingHandoffs.add(Futures.transform(
+    pendingHandoffs.add(Futures.transformAsync(
         publishFuture,
-        (AsyncFunction<SegmentsAndCommitMetadata, SegmentsAndCommitMetadata>) driver::registerHandoff
+	(AsyncFunction<SegmentsAndCommitMetadata, SegmentsAndCommitMetadata>) driver::registerHandoff,
+        MoreExecutors.directExecutor()
     ));
   }
 

diff --git a/...n/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexPhaseRunner.java b/...n/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexPhaseRunner.java
@@ -24,6 +24,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.druid.indexer.TaskState;
 import org.apache.druid.indexer.TaskStatusPlus;
 import org.apache.druid.indexing.common.TaskToolbox;
@@ -262,7 +263,8 @@ public void onFailure(Throwable t)
             LOG.error(t, "Error while running a task for spec[%s]", spec.getId());
             taskCompleteEvents.offer(SubTaskCompleteEvent.fail(spec, t));
           }
-        }
+        },
+        MoreExecutors.directExecutor()
     );
   }
 

diff --git a/...-service/src/main/java/org/apache/druid/indexing/overlord/QuotableWhiteSpaceSplitter.java b/...-service/src/main/java/org/apache/druid/indexing/overlord/QuotableWhiteSpaceSplitter.java
@@ -54,7 +54,7 @@ public boolean matches(char c)
             if (inQuotes) {
               return false;
             }
-            return CharMatcher.BREAKING_WHITESPACE.matches(c);
+            return CharMatcher.breakingWhitespace().matches(c);
           }
         }
     ).omitEmptyStrings().split(string).iterator();

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/RemoteTaskRunner.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/RemoteTaskRunner.java
@@ -266,7 +266,8 @@ public void onFailure(Throwable throwable)
                           waitingForMonitor.notifyAll();
                         }
                       }
-                    }
+                    },
+                    MoreExecutors.directExecutor()
                 );
                 break;
               case CHILD_UPDATED:
@@ -1298,7 +1299,8 @@ public void onFailure(Throwable t)
           {
             removedWorkerCleanups.remove(worker, cleanupTask);
           }
-        }
+        },
+        MoreExecutors.directExecutor()
     );
   }
 

diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskLockbox.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/TaskLockbox.java
@@ -21,6 +21,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Objects;
+import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ComparisonChain;
 import com.google.common.collect.ImmutableList;
@@ -1560,10 +1561,10 @@ public int hashCode()
     @Override
     public String toString()
     {
-      return Objects.toStringHelper(this)
-                    .add("taskLock", taskLock)
-                    .add("taskIds", taskIds)
-                    .toString();
+      return MoreObjects.toStringHelper(this)
+                    	.add("taskLock", taskLock)
+                    	.add("taskIds", taskIds)
+                    	.toString();
     }
   }
 

diff --git a/...g-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java b/...g-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java
@@ -757,7 +757,8 @@ public void onFailure(Throwable t)
           {
             removedWorkerCleanups.remove(workerHostAndPort, cleanupTask);
           }
-        }
+        },
+        MoreExecutors.directExecutor()
     );
   }
 

diff --git a/...java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClientAsyncImpl.java b/...java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClientAsyncImpl.java
@@ -26,6 +26,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import org.apache.druid.common.guava.FutureUtils;
 import org.apache.druid.indexer.TaskLocation;
@@ -397,7 +398,8 @@ public void onFailure(Throwable t)
                             {
                               retVal.setException(t);
                             }
-                          }
+                          },
+                          MoreExecutors.directExecutor()
                       ),
                   sleepTime,
                   TimeUnit.MILLISECONDS
@@ -574,7 +576,8 @@ public void onFailure(Throwable t)
                 retVal.set(either.valueOrThrow());
               }
             }
-          }
+          },
+           MoreExecutors.directExecutor()
       );
 
       return retVal;

diff --git a/...src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java b/...src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
@@ -36,6 +36,7 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import org.apache.druid.data.input.Committer;
 import org.apache.druid.data.input.InputFormat;
 import org.apache.druid.data.input.InputRow;
@@ -137,6 +138,7 @@
  * @param <PartitionIdType>    Partition Number Type
  * @param <SequenceOffsetType> Sequence Number Type
  */
+@SuppressWarnings("CheckReturnValue")
 public abstract class SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOffsetType, RecordType extends ByteEntity> implements ChatHandler
 {
   public enum Status
@@ -697,7 +699,8 @@ public void onFailure(Throwable t)
                         log.error("Persist failed, dying");
                         backgroundThreadException = t;
                       }
-                    }
+                    },
+                    MoreExecutors.directExecutor()
                 );
               }
 
@@ -970,7 +973,8 @@ private void publishAndRegisterHandoff(SequenceMetadata<PartitionIdType, Sequenc
           } else {
             return publishedSegmentsAndMetadata;
           }
-        }
+        },
+         MoreExecutors.directExecutor()
     );
     publishWaitList.add(publishFuture);
 
@@ -1026,7 +1030,8 @@ public Void apply(@Nullable SegmentsAndCommitMetadata handoffSegmentsAndCommitMe
                     handoffFuture.set(handoffSegmentsAndCommitMetadata);
                     return null;
                   }
-                }
+                },
+                    MoreExecutors.directExecutor()
             );
             // emit segment count metric:
             int segmentCount = 0;
@@ -1047,7 +1052,8 @@ public void onFailure(Throwable t)
             log.error(t, "Error while publishing segments for sequenceNumber[%s]", sequenceMetadata);
             handoffFuture.setException(t);
           }
-        }
+        },
+                    MoreExecutors.directExecutor()
     );
   }
-Original file line number
+Diff line change
@@ Expand Up / @@ -1199,7 +1199,7 @@ public void onFailure(Throwable t) @@
                 settableFuture.setException(t);
               }
             }
-          });
+          }, MoreExecutors.directExecutor());
           taskFutures.add(settableFuture);
         }
@@ Expand Down @@