integration test for coordinator and overlord leadership client

.travis.yml

@@ -483,13 +483,19 @@ jobs:
       name: "(Compile=openjdk8, Run=openjdk8) other integration tests"
       jdk: openjdk8
       services: *integration_test_services
-      env: TESTNG_GROUPS='-DexcludedGroups=batch-index,input-format,input-source,perfect-rollup-parallel-batch-index,kafka-index,query,query-retry,realtime-index,security,s3-deep-storage,gcs-deep-storage,azure-deep-storage,hdfs-deep-storage,s3-ingestion,kinesis-index,kinesis-data-format,kafka-transactional-index,kafka-index-slow,kafka-transactional-index-slow,kafka-data-format,hadoop-s3-to-s3-deep-storage,hadoop-s3-to-hdfs-deep-storage,hadoop-azure-to-azure-deep-storage,hadoop-azure-to-hdfs-deep-storage,hadoop-gcs-to-gcs-deep-storage,hadoop-gcs-to-hdfs-deep-storage,aliyun-oss-deep-storage,append-ingestion,compaction' JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='middleManager'
+      env: TESTNG_GROUPS='-DexcludedGroups=batch-index,input-format,input-source,perfect-rollup-parallel-batch-index,kafka-index,query,query-retry,realtime-index,security,s3-deep-storage,gcs-deep-storage,azure-deep-storage,hdfs-deep-storage,s3-ingestion,kinesis-index,kinesis-data-format,kafka-transactional-index,kafka-index-slow,kafka-transactional-index-slow,kafka-data-format,hadoop-s3-to-s3-deep-storage,hadoop-s3-to-hdfs-deep-storage,hadoop-azure-to-azure-deep-storage,hadoop-azure-to-hdfs-deep-storage,hadoop-gcs-to-gcs-deep-storage,hadoop-gcs-to-hdfs-deep-storage,aliyun-oss-deep-storage,append-ingestion,compaction,high-availability' JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='middleManager'
       script: *run_integration_test
       after_failure: *integration_test_diags
 
     - <<: *integration_tests
       name: "(Compile=openjdk8, Run=openjdk8) other integration tests with Indexer"
-      env: TESTNG_GROUPS='-DexcludedGroups=batch-index,input-format,input-source,perfect-rollup-parallel-batch-index,kafka-index,query,query-retry,realtime-index,security,s3-deep-storage,gcs-deep-storage,azure-deep-storage,hdfs-deep-storage,s3-ingestion,kinesis-index,kinesis-data-format,kafka-transactional-index,kafka-index-slow,kafka-transactional-index-slow,kafka-data-format,hadoop-s3-to-s3-deep-storage,hadoop-s3-to-hdfs-deep-storage,hadoop-azure-to-azure-deep-storage,hadoop-azure-to-hdfs-deep-storage,hadoop-gcs-to-gcs-deep-storage,hadoop-gcs-to-hdfs-deep-storage,aliyun-oss-deep-storage,append-ingestion,compaction' JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='indexer'
+      env: TESTNG_GROUPS='-DexcludedGroups=batch-index,input-format,input-source,perfect-rollup-parallel-batch-index,kafka-index,query,query-retry,realtime-index,security,s3-deep-storage,gcs-deep-storage,azure-deep-storage,hdfs-deep-storage,s3-ingestion,kinesis-index,kinesis-data-format,kafka-transactional-index,kafka-index-slow,kafka-transactional-index-slow,kafka-data-format,hadoop-s3-to-s3-deep-storage,hadoop-s3-to-hdfs-deep-storage,hadoop-azure-to-azure-deep-storage,hadoop-azure-to-hdfs-deep-storage,hadoop-gcs-to-gcs-deep-storage,hadoop-gcs-to-hdfs-deep-storage,aliyun-oss-deep-storage,append-ingestion,compaction,high-availability' JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='indexer'
+
+    - <<: *integration_tests
+      name: "(Compile=openjdk8, Run=openjdk8) leadership and high availability integration tests"
+      jdk: openjdk8
+      env: TESTNG_GROUPS='-Dgroups=high-availability' JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='middleManager'
+
     # END - Integration tests for Compile with Java 8 and Run with Java 8
 
     # START - Integration tests for Compile with Java 8 and Run with Java 11
@@ -546,7 +552,12 @@ jobs:
     - <<: *integration_tests
       name: "(Compile=openjdk8, Run=openjdk11) other integration test"
       jdk: openjdk8
-      env: TESTNG_GROUPS='-DexcludedGroups=batch-index,input-format,input-source,perfect-rollup-parallel-batch-index,kafka-index,query,query-retry,realtime-index,security,s3-deep-storage,gcs-deep-storage,azure-deep-storage,hdfs-deep-storage,s3-ingestion,kinesis-index,kinesis-data-format,kafka-transactional-index,kafka-index-slow,kafka-transactional-index-slow,kafka-data-format,hadoop-s3-to-s3-deep-storage,hadoop-s3-to-hdfs-deep-storage,hadoop-azure-to-azure-deep-storage,hadoop-azure-to-hdfs-deep-storage,hadoop-gcs-to-gcs-deep-storage,hadoop-gcs-to-hdfs-deep-storage,aliyun-oss-deep-storage,append-ingestion,compaction' JVM_RUNTIME='-Djvm.runtime=11' USE_INDEXER='middleManager'
+      env: TESTNG_GROUPS='-DexcludedGroups=batch-index,input-format,input-source,perfect-rollup-parallel-batch-index,kafka-index,query,query-retry,realtime-index,security,s3-deep-storage,gcs-deep-storage,azure-deep-storage,hdfs-deep-storage,s3-ingestion,kinesis-index,kinesis-data-format,kafka-transactional-index,kafka-index-slow,kafka-transactional-index-slow,kafka-data-format,hadoop-s3-to-s3-deep-storage,hadoop-s3-to-hdfs-deep-storage,hadoop-azure-to-azure-deep-storage,hadoop-azure-to-hdfs-deep-storage,hadoop-gcs-to-gcs-deep-storage,hadoop-gcs-to-hdfs-deep-storage,aliyun-oss-deep-storage,append-ingestion,compaction,high-availability' JVM_RUNTIME='-Djvm.runtime=11' USE_INDEXER='middleManager'
+
+    - <<: *integration_tests
+      name: "(Compile=openjdk8, Run=openjdk11) leadership and high availability integration tests"
+      jdk: openjdk8
+      env: TESTNG_GROUPS='-Dgroups=high-availability' JVM_RUNTIME='-Djvm.runtime=11' USE_INDEXER='middleManager'
     # END - Integration tests for Compile with Java 8 and Run with Java 11
 
     - name: "security vulnerabilities"

docs/querying/sql.md

@@ -1139,6 +1139,7 @@ Servers table lists all discovered servers in the cluster.
 |tier|STRING|Distribution tier see [druid.server.tier](../configuration/index.html#historical-general-configuration). Only valid for HISTORICAL type, for other types it's null|
 |current_size|LONG|Current size of segments in bytes on this server. Only valid for HISTORICAL type, for other types it's 0|
 |max_size|LONG|Max size in bytes this server recommends to assign to segments see [druid.server.maxSize](../configuration/index.html#historical-general-configuration). Only valid for HISTORICAL type, for other types it's 0|
+|is_leader|LONG|1 if the server is currently the 'leader' (for services which have the concept of leadership), otherwise 0 if the server is not the leader, or the default long value (0 or null depending on `druid.generic.useDefaultValueForNull`) if the server type does not have the concept of leadership|
 
 To retrieve information about all servers, use the query:
 

extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/BasicAuthUtils.java

@@ -21,8 +21,10 @@
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Predicate;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.RetryUtils;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.security.basic.authentication.entity.BasicAuthenticatorUser;
@@ -66,6 +68,9 @@ public class BasicAuthUtils
   public static final int DEFAULT_CREDENTIAL_CACHE_SIZE = 100;
   public static final int KEY_LENGTH = 512;
   public static final String ALGORITHM = "PBKDF2WithHmacSHA512";
+  public static final int MAX_INIT_RETRIES = 2;
+  public static final Predicate<Throwable> SHOULD_RETRY_INIT =
+      (throwable) -> throwable instanceof BasicSecurityDBResourceException;
 
   public static final TypeReference<Map<String, BasicAuthenticatorUser>> AUTHENTICATOR_USER_MAP_TYPE_REFERENCE =
       new TypeReference<Map<String, BasicAuthenticatorUser>>()
@@ -277,4 +282,14 @@ public static byte[] serializeAuthorizerRoleMap(ObjectMapper objectMapper, Map<S
       throw new ISE(ioe, "Couldn't serialize authorizer roleMap!");
     }
   }
+
+  public static void maybeInitialize(final RetryUtils.Task<?> task)
+  {
+    try {
+      RetryUtils.retry(task, SHOULD_RETRY_INIT, MAX_INIT_RETRIES);
+    }
+    catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
 }

extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authentication/db/updater/CoordinatorBasicAuthenticatorMetadataStorageUpdater.java

@@ -119,46 +119,50 @@ public void start()
 
     try {
       LOG.info("Starting CoordinatorBasicAuthenticatorMetadataStorageUpdater.");
-      for (Map.Entry<String, Authenticator> entry : authenticatorMapper.getAuthenticatorMap().entrySet()) {
-        Authenticator authenticator = entry.getValue();
-        if (authenticator instanceof BasicHTTPAuthenticator) {
-          String authenticatorName = entry.getKey();
-          authenticatorPrefixes.add(authenticatorName);
-          BasicHTTPAuthenticator basicHTTPAuthenticator = (BasicHTTPAuthenticator) authenticator;
-          BasicAuthDBConfig dbConfig = basicHTTPAuthenticator.getDbConfig();
-          byte[] userMapBytes = getCurrentUserMapBytes(authenticatorName);
-          Map<String, BasicAuthenticatorUser> userMap = BasicAuthUtils.deserializeAuthenticatorUserMap(
-              objectMapper,
-              userMapBytes
-          );
-          cachedUserMaps.put(authenticatorName, new BasicAuthenticatorUserMapBundle(userMap, userMapBytes));
-
-          if (dbConfig.getInitialAdminPassword() != null && !userMap.containsKey(BasicAuthUtils.ADMIN_NAME)) {
-            createUserInternal(authenticatorName, BasicAuthUtils.ADMIN_NAME);
-            setUserCredentialsInternal(
-                authenticatorName,
-                BasicAuthUtils.ADMIN_NAME,
-                new BasicAuthenticatorCredentialUpdate(
-                    dbConfig.getInitialAdminPassword().getPassword(),
-                    BasicAuthUtils.DEFAULT_KEY_ITERATIONS
-                )
-            );
-          }
+      BasicAuthUtils.maybeInitialize(
+          () -> {
+            for (Map.Entry<String, Authenticator> entry : authenticatorMapper.getAuthenticatorMap().entrySet()) {
+              Authenticator authenticator = entry.getValue();
+              if (authenticator instanceof BasicHTTPAuthenticator) {
+                String authenticatorName = entry.getKey();
+                authenticatorPrefixes.add(authenticatorName);
+                BasicHTTPAuthenticator basicHTTPAuthenticator = (BasicHTTPAuthenticator) authenticator;
+                BasicAuthDBConfig dbConfig = basicHTTPAuthenticator.getDbConfig();
+                byte[] userMapBytes = getCurrentUserMapBytes(authenticatorName);
+                Map<String, BasicAuthenticatorUser> userMap = BasicAuthUtils.deserializeAuthenticatorUserMap(
+                    objectMapper,
+                    userMapBytes
+                );
+                cachedUserMaps.put(authenticatorName, new BasicAuthenticatorUserMapBundle(userMap, userMapBytes));
+
+                if (dbConfig.getInitialAdminPassword() != null && !userMap.containsKey(BasicAuthUtils.ADMIN_NAME)) {
+                  createUserInternal(authenticatorName, BasicAuthUtils.ADMIN_NAME);
+                  setUserCredentialsInternal(
+                      authenticatorName,
+                      BasicAuthUtils.ADMIN_NAME,
+                      new BasicAuthenticatorCredentialUpdate(
+                          dbConfig.getInitialAdminPassword().getPassword(),
+                          BasicAuthUtils.DEFAULT_KEY_ITERATIONS
+                      )
+                  );
+                }
 
-          if (dbConfig.getInitialInternalClientPassword() != null
-              && !userMap.containsKey(BasicAuthUtils.INTERNAL_USER_NAME)) {
-            createUserInternal(authenticatorName, BasicAuthUtils.INTERNAL_USER_NAME);
-            setUserCredentialsInternal(
-                authenticatorName,
-                BasicAuthUtils.INTERNAL_USER_NAME,
-                new BasicAuthenticatorCredentialUpdate(
-                    dbConfig.getInitialInternalClientPassword().getPassword(),
-                    BasicAuthUtils.DEFAULT_KEY_ITERATIONS
-                )
-            );
-          }
-        }
-      }
+                if (dbConfig.getInitialInternalClientPassword() != null
+                    && !userMap.containsKey(BasicAuthUtils.INTERNAL_USER_NAME)) {
+                  createUserInternal(authenticatorName, BasicAuthUtils.INTERNAL_USER_NAME);
+                  setUserCredentialsInternal(
+                      authenticatorName,
+                      BasicAuthUtils.INTERNAL_USER_NAME,
+                      new BasicAuthenticatorCredentialUpdate(
+                          dbConfig.getInitialInternalClientPassword().getPassword(),
+                          BasicAuthUtils.DEFAULT_KEY_ITERATIONS
+                      )
+                  );
+                }
+              }
+            }
+            return true;
+          });
 
       ScheduledExecutors.scheduleWithFixedDelay(
           exec,

extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/authorization/db/updater/CoordinatorBasicAuthorizerMetadataStorageUpdater.java

@@ -143,42 +143,52 @@ public void start()
 
     try {
       LOG.info("Starting CoordinatorBasicAuthorizerMetadataStorageUpdater");
-      for (Map.Entry<String, Authorizer> entry : authorizerMapper.getAuthorizerMap().entrySet()) {
-        Authorizer authorizer = entry.getValue();
-        if (authorizer instanceof BasicRoleBasedAuthorizer) {
-          BasicRoleBasedAuthorizer basicRoleBasedAuthorizer = (BasicRoleBasedAuthorizer) authorizer;
-          BasicAuthDBConfig dbConfig = basicRoleBasedAuthorizer.getDbConfig();
-          String authorizerName = entry.getKey();
-          authorizerNames.add(authorizerName);
-
-          byte[] userMapBytes = getCurrentUserMapBytes(authorizerName);
-          Map<String, BasicAuthorizerUser> userMap = BasicAuthUtils.deserializeAuthorizerUserMap(
-              objectMapper,
-              userMapBytes
-          );
-          cachedUserMaps.put(authorizerName, new BasicAuthorizerUserMapBundle(userMap, userMapBytes));
+      BasicAuthUtils.maybeInitialize(
+          () -> {
+            for (Map.Entry<String, Authorizer> entry : authorizerMapper.getAuthorizerMap().entrySet()) {
+              Authorizer authorizer = entry.getValue();
+              if (authorizer instanceof BasicRoleBasedAuthorizer) {
+                BasicRoleBasedAuthorizer basicRoleBasedAuthorizer = (BasicRoleBasedAuthorizer) authorizer;
+                BasicAuthDBConfig dbConfig = basicRoleBasedAuthorizer.getDbConfig();
+                String authorizerName = entry.getKey();
+                authorizerNames.add(authorizerName);
 
-          byte[] groupMappingMapBytes = getCurrentGroupMappingMapBytes(authorizerName);
-          Map<String, BasicAuthorizerGroupMapping> groupMappingMap = BasicAuthUtils.deserializeAuthorizerGroupMappingMap(
-              objectMapper,
-              groupMappingMapBytes
-          );
-          cachedGroupMappingMaps.put(authorizerName, new BasicAuthorizerGroupMappingMapBundle(groupMappingMap, groupMappingMapBytes));
+                byte[] userMapBytes = getCurrentUserMapBytes(authorizerName);
+                Map<String, BasicAuthorizerUser> userMap = BasicAuthUtils.deserializeAuthorizerUserMap(
+                    objectMapper,
+                    userMapBytes
+                );
+                cachedUserMaps.put(authorizerName, new BasicAuthorizerUserMapBundle(userMap, userMapBytes));
 
-          byte[] roleMapBytes = getCurrentRoleMapBytes(authorizerName);
-          Map<String, BasicAuthorizerRole> roleMap = BasicAuthUtils.deserializeAuthorizerRoleMap(
-              objectMapper,
-              roleMapBytes
-          );
-          cachedRoleMaps.put(authorizerName, new BasicAuthorizerRoleMapBundle(roleMap, roleMapBytes));
+                byte[] groupMappingMapBytes = getCurrentGroupMappingMapBytes(authorizerName);
+                Map<String, BasicAuthorizerGroupMapping> groupMappingMap = BasicAuthUtils.deserializeAuthorizerGroupMappingMap(
+                    objectMapper,
+                    groupMappingMapBytes
+                );
+                cachedGroupMappingMaps.put(
+                    authorizerName,
+                    new BasicAuthorizerGroupMappingMapBundle(
+                        groupMappingMap,
+                        groupMappingMapBytes
+                    )
+                );
 
-          initSuperUsersAndGroupMapping(authorizerName, userMap, roleMap, groupMappingMap,
-                                        dbConfig.getInitialAdminUser(),
-                                        dbConfig.getInitialAdminRole(),
-                                        dbConfig.getInitialAdminGroupMapping()
-          );
-        }
-      }
+                byte[] roleMapBytes = getCurrentRoleMapBytes(authorizerName);
+                Map<String, BasicAuthorizerRole> roleMap = BasicAuthUtils.deserializeAuthorizerRoleMap(
+                    objectMapper,
+                    roleMapBytes
+                );
+                cachedRoleMaps.put(authorizerName, new BasicAuthorizerRoleMapBundle(roleMap, roleMapBytes));
+
+                initSuperUsersAndGroupMapping(authorizerName, userMap, roleMap, groupMappingMap,
+                                              dbConfig.getInitialAdminUser(),
+                                              dbConfig.getInitialAdminRole(),
+                                              dbConfig.getInitialAdminGroupMapping()
+                );
+              }
+            }
+            return true;
+          });
 
       ScheduledExecutors.scheduleWithFixedDelay(
           exec,

indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java

@@ -484,6 +484,7 @@ public TaskStatusResponse getTaskStatus(String taskId)
       final String groupId = task.isPresent() ? task.get().getGroupId() : null;
       final String taskType = task.isPresent() ? task.get().getType() : null;
       final TaskStatus taskStatus = taskRunner.getStatus(taskId);
+
       if (taskStatus != null) {
         return new TaskStatusResponse(
             taskId,