Fix empty datasource schema on the Broker when metadata query is disabled (#16645)

* Fix build

* Fix empty datasource schema on the broker

* review comment

* Remove unused import
This commit is contained in:
Rishabh Singh 2024-06-28 11:06:56 +05:30 committed by GitHub
parent 45c020060c
commit b9c7664ac3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 62 additions and 1 deletions

View File

@ -81,7 +81,7 @@ services:
service: druid-broker
environment:
- DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
- druid_sql_planner_metadataRefreshPeriod=PT20S
- druid_sql_planner_metadataRefreshPeriod=PT30S
- druid_sql_planner_disableSegmentMetadataQueries=true
depends_on:
- druid-coordinator

View File

@ -23,6 +23,7 @@ import com.google.inject.Inject;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
import org.testng.Assert;
public final class DataLoaderHelper
{
@ -50,6 +51,9 @@ public final class DataLoaderHelper
() -> sqlTestQueryHelper.isDatasourceLoadedInSQL(datasource),
StringUtils.format("Waiting for [%s] to be ready for SQL queries", datasource)
);
Assert.assertTrue(sqlTestQueryHelper.verifyTimeColumnIsPresent(datasource));
LOG.info("Datasource [%s] ready for SQL queries", datasource);
}
}

View File

@ -69,4 +69,27 @@ public class SqlTestQueryHelper extends AbstractTestQueryHelper<SqlQueryWithResu
return false;
}
}
public boolean verifyTimeColumnIsPresent(String datasource)
{
final SqlQuery query = new SqlQuery(
"SELECT __time FROM \"" + datasource + "\" LIMIT 1",
null,
false,
false,
false,
null,
null
);
try {
//noinspection unchecked
queryClient.query(getQueryURL(broker), query);
return true;
}
catch (Exception e) {
LOG.debug(e, "Check query failed");
return false;
}
}
}

View File

@ -246,6 +246,16 @@ public class BrokerSegmentMetadataCache extends AbstractSegmentMetadataCache<Phy
continue;
}
if (rowSignature.getColumnNames().isEmpty()) {
// this case could arise when metadata refresh is disabled on broker
// and a new datasource is added
log.info("datasource [%s] schema has not been initialized yet, "
+ "check coordinator logs if this message is persistent.", dataSource);
// this is a harmless call
tables.remove(dataSource);
continue;
}
final PhysicalDatasourceMetadata physicalDatasourceMetadata = dataSourceMetadataFactory.build(dataSource, rowSignature);
updateDSMetadata(dataSource, physicalDatasourceMetadata);
}

View File

@ -1027,4 +1027,28 @@ public class BrokerSegmentMetadataCacheTest extends BrokerSegmentMetadataCacheTe
buildSchemaMarkAndTableLatch();
serverView.invokeSegmentSchemasAnnouncedDummy();
}
@Test
public void testNoDatasourceSchemaWhenNoSegmentMetadata() throws InterruptedException, IOException
{
BrokerSegmentMetadataCacheConfig config = new BrokerSegmentMetadataCacheConfig();
config.setDisableSegmentMetadataQueries(true);
BrokerSegmentMetadataCache schema = buildSchemaMarkAndTableLatch(
config,
new NoopCoordinatorClient()
);
schema.start();
schema.awaitInitialization();
List<DataSegment> segments = schema.getSegmentMetadataSnapshot().values()
.stream()
.map(AvailableSegmentMetadata::getSegment)
.collect(Collectors.toList());
schema.refresh(segments.stream().map(DataSegment::getId).collect(Collectors.toSet()), Collections.singleton("foo"));
Assert.assertNull(schema.getDatasource("foo"));
}
}