Load broadcast datasources on broker and tasks

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java

@@ -191,4 +191,10 @@ public String getType()
   {
     return TYPE;
   }
+
+  @Override
+  public boolean supportsQueries()
+  {
+    return true;
+  }
 }

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -344,6 +344,7 @@ public void testRunAfterDataInserted() throws Exception
             INPUT_FORMAT
         )
     );
+    Assert.assertTrue(task.supportsQueries());
 
     final ListenableFuture<TaskStatus> future = runTask(task);
 

extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java

@@ -137,6 +137,12 @@ public String getType()
     return TYPE;
   }
 
+  @Override
+  public boolean supportsQueries()
+  {
+    return true;
+  }
+
   @VisibleForTesting
   AWSCredentialsConfig getAwsCredentialsConfig()
   {

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java

@@ -346,8 +346,8 @@ public void testRunAfterDataInserted() throws Exception
             null,
             false
         )
-
     );
+    Assert.assertTrue(task.supportsQueries());
 
     final ListenableFuture<TaskStatus> future = runTask(task);
 

indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractTask.java

@@ -145,6 +145,12 @@ public <T> QueryRunner<T> getQueryRunner(Query<T> query)
     return null;
   }
 
+  @Override
+  public boolean supportsQueries()
+  {
+    return false;
+  }
+
   @Override
   public String getClasspathPrefix()
   {

indexing-service/src/main/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTask.java

@@ -245,6 +245,12 @@ public <T> QueryRunner<T> getQueryRunner(Query<T> query)
     return (queryPlus, responseContext) -> queryPlus.run(appenderator, responseContext);
   }
 
+  @Override
+  public boolean supportsQueries()
+  {
+    return true;
+  }
+
   @Override
   public boolean isReady(TaskActionClient taskActionClient)
   {

indexing-service/src/main/java/org/apache/druid/indexing/common/task/RealtimeIndexTask.java

@@ -196,6 +196,12 @@ public <T> QueryRunner<T> getQueryRunner(Query<T> query)
     return plumber.getQueryRunner(query);
   }
 
+  @Override
+  public boolean supportsQueries()
+  {
+    return true;
+  }
+
   @Override
   public boolean isReady(TaskActionClient taskActionClient)
   {

indexing-service/src/main/java/org/apache/druid/indexing/common/task/Task.java

@@ -146,6 +146,11 @@ default int getPriority()
    */
   <T> QueryRunner<T> getQueryRunner(Query<T> query);
 
+  /**
+   * @return true if this Task type is queryable, such as streaming ingestion tasks
+   */
+  boolean supportsQueries();
+
   /**
    * Returns an extra classpath that should be prepended to the default classpath when running this task. If no
    * extra classpath should be prepended, this should return null or the empty string.

indexing-service/src/main/java/org/apache/druid/indexing/overlord/ForkingTaskRunner.java

@@ -327,6 +327,13 @@ public TaskStatus call()
                           command.add(nodeType);
                         }
 
+                        // If the task type is queryable, we need to load broadcast segments on the peon, used for
+                        // join queries
+                        if (task.supportsQueries()) {
+                          command.add("--loadBroadcastSegments");
+                          command.add("true");
+                        }
+
                         if (!taskFile.exists()) {
                           jsonMapper.writeValue(taskFile, task);
                         }

indexing-service/src/test/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java

@@ -337,6 +337,7 @@ public void testBasics() throws Exception
   {
     expectPublishedSegments(1);
     final AppenderatorDriverRealtimeIndexTask task = makeRealtimeTask(null);
+    Assert.assertTrue(task.supportsQueries());
     final ListenableFuture<TaskStatus> statusFuture = runTask(task);
 
     // Wait for firehose to show up, it starts off null.

indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java

@@ -229,6 +229,8 @@ public void testDeterminePartitions() throws Exception
         appenderatorsManager
     );
 
+    Assert.assertFalse(indexTask.supportsQueries());
+
     final List<DataSegment> segments = runTask(indexTask).rhs;
 
     Assert.assertEquals(2, segments.size());

indexing-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java

@@ -189,6 +189,12 @@ public void testDefaultResource()
     Assert.assertEquals(task.getId(), task.getTaskResource().getAvailabilityGroup());
   }
 
+  @Test(timeout = 60_000L)
+  public void testSupportsQueries()
+  {
+    final RealtimeIndexTask task = makeRealtimeTask(null);
+    Assert.assertTrue(task.supportsQueries());
+  }
 
   @Test(timeout = 60_000L, expected = ExecutionException.class)
   public void testHandoffTimeout() throws Exception

server/pom.xml

@@ -291,10 +291,6 @@
             <groupId>javax.validation</groupId>
             <artifactId>validation-api</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.hibernate</groupId>
-            <artifactId>hibernate-validator</artifactId>
-        </dependency>
         <dependency>
             <groupId>com.google.errorprone</groupId>
             <artifactId>error_prone_annotations</artifactId>

server/src/main/java/org/apache/druid/client/CachingClusteredClient.java

@@ -452,7 +452,7 @@ private String computeCurrentEtag(final Set<SegmentServerSelector> segments, @Nu
       Hasher hasher = Hashing.sha1().newHasher();
       boolean hasOnlyHistoricalSegments = true;
       for (SegmentServerSelector p : segments) {
-        if (!p.getServer().pick().getServer().segmentReplicatable()) {
+        if (!p.getServer().pick().getServer().isSegmentReplicationTarget()) {
           hasOnlyHistoricalSegments = false;
           break;
         }
@@ -633,7 +633,7 @@ private void addSequencesFromServer(
 
         if (isBySegment) {
           serverResults = getBySegmentServerResults(serverRunner, segmentsOfServer, maxQueuedBytesPerServer);
-        } else if (!server.segmentReplicatable() || !populateCache) {
+        } else if (!server.isSegmentReplicationTarget() || !populateCache) {
           serverResults = getSimpleServerResults(serverRunner, segmentsOfServer, maxQueuedBytesPerServer);
         } else {
           serverResults = getAndCacheServerResults(serverRunner, segmentsOfServer, maxQueuedBytesPerServer);

server/src/main/java/org/apache/druid/client/DruidServer.java

@@ -137,9 +137,19 @@ public String getTier()
     return metadata.getTier();
   }
 
-  public boolean segmentReplicatable()
+  public boolean isSegmentReplicationTarget()
   {
-    return metadata.segmentReplicatable();
+    return metadata.isSegmentReplicationTarget();
+  }
+
+  public boolean isSegmentBroadcastTarget()
+  {
+    return metadata.isSegmentBroadcastTarget();
+  }
+
+  public boolean isSegmentReplicationOrBroadcastTarget()
+  {
+    return metadata.isSegmentReplicationTarget() || metadata.isSegmentBroadcastTarget();
   }
 
   @JsonProperty

server/src/main/java/org/apache/druid/discovery/DruidNodeDiscoveryProvider.java

@@ -44,7 +44,8 @@ public abstract class DruidNodeDiscoveryProvider
   private static final Map<String, Set<NodeRole>> SERVICE_TO_NODE_TYPES = ImmutableMap.of(
       LookupNodeService.DISCOVERY_SERVICE_KEY,
       ImmutableSet.of(NodeRole.BROKER, NodeRole.HISTORICAL, NodeRole.PEON, NodeRole.INDEXER),
-      DataNodeService.DISCOVERY_SERVICE_KEY, ImmutableSet.of(NodeRole.HISTORICAL, NodeRole.PEON, NodeRole.INDEXER),
+      DataNodeService.DISCOVERY_SERVICE_KEY,
+      ImmutableSet.of(NodeRole.HISTORICAL, NodeRole.PEON, NodeRole.INDEXER, NodeRole.BROKER),
       WorkerNodeService.DISCOVERY_SERVICE_KEY, ImmutableSet.of(NodeRole.MIDDLE_MANAGER, NodeRole.INDEXER)
   );
 

server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderConfig.java

@@ -23,9 +23,9 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import org.apache.druid.utils.JvmUtils;
-import org.hibernate.validator.constraints.NotEmpty;
 
 import java.io.File;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
@@ -34,8 +34,7 @@
 public class SegmentLoaderConfig
 {
   @JsonProperty
-  @NotEmpty
-  private List<StorageLocationConfig> locations = null;
+  private List<StorageLocationConfig> locations = Collections.emptyList();
 
   @JsonProperty("lazyLoadOnStart")
   private boolean lazyLoadOnStart = false;

server/src/main/java/org/apache/druid/segment/loading/SegmentLoaderLocalCacheManager.java

@@ -89,7 +89,6 @@ public SegmentLoaderLocalCacheManager(
     this.indexIO = indexIO;
     this.config = config;
     this.jsonMapper = mapper;
-
     this.locations = new ArrayList<>();
     for (StorageLocationConfig locationConfig : config.getLocations()) {
       locations.add(

server/src/main/java/org/apache/druid/segment/realtime/plumber/CoordinatorBasedSegmentHandoffNotifier.java

@@ -142,7 +142,7 @@ static boolean isHandOffComplete(List<ImmutableSegmentLoadInfo> serverView, Segm
           && segmentLoadInfo.getSegment().getShardSpec().getPartitionNum()
              == descriptor.getPartitionNumber()
           && segmentLoadInfo.getSegment().getVersion().compareTo(descriptor.getVersion()) >= 0
-          && segmentLoadInfo.getServers().stream().anyMatch(DruidServerMetadata::segmentReplicatable)) {
+          && segmentLoadInfo.getServers().stream().anyMatch(DruidServerMetadata::isSegmentReplicationOrBroadcastTarget)) {
         return true;
       }
     }

server/src/main/java/org/apache/druid/server/coordination/DruidServerMetadata.java

@@ -107,11 +107,21 @@ public int getPriority()
     return priority;
   }
 
-  public boolean segmentReplicatable()
+  public boolean isSegmentReplicationTarget()
   {
     return type.isSegmentReplicationTarget();
   }
 
+  public boolean isSegmentBroadcastTarget()
+  {
+    return type.isSegmentBroadcastTarget();
+  }
+
+  public boolean isSegmentReplicationOrBroadcastTarget()
+  {
+    return isSegmentReplicationTarget() || isSegmentBroadcastTarget();
+  }
+
   @Override
   public boolean equals(Object o)
   {