You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by gi...@apache.org on 2025/05/13 15:43:30 UTC
(druid) branch master updated: Fix Coordinator API MetadataResource.getSegment (#17998)
This is an automated email from the ASF dual-hosted git repository.
gian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 80ae9c36f01 Fix Coordinator API MetadataResource.getSegment (#17998)
80ae9c36f01 is described below
commit 80ae9c36f01605dc1784b83363ba052305a3924e
Author: Kashif Faraz <ka...@gmail.com>
AuthorDate: Tue May 13 21:13:24 2025 +0530
Fix Coordinator API MetadataResource.getSegment (#17998)
---
.../common/actions/SegmentAllocateActionTest.java | 2 +-
.../AbstractParallelIndexSupervisorTaskTest.java | 7 +-
.../TestIndexerMetadataStorageCoordinator.java | 6 +-
.../IndexerMetadataStorageCoordinator.java | 6 +-
.../IndexerSQLMetadataStorageCoordinator.java | 24 +-
.../apache/druid/server/http/MetadataResource.java | 24 +-
...rSQLMetadataStorageCoordinatorReadOnlyTest.java | 320 +++++++++++++++++++++
.../IndexerSQLMetadataStorageCoordinatorTest.java | 9 +-
.../druid/server/http/MetadataResourceTest.java | 6 +-
9 files changed, 367 insertions(+), 37 deletions(-)
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentAllocateActionTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentAllocateActionTest.java
index be5f6c3ab39..66fb89456e9 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentAllocateActionTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentAllocateActionTest.java
@@ -1173,7 +1173,7 @@ public class SegmentAllocateActionTest
// Allocate another id and ensure that it doesn't exist in the druid_segments table
final SegmentIdWithShardSpec theId =
allocate(task1, DateTimes.nowUtc(), Granularities.NONE, Granularities.ALL, "seq", "3");
- Assert.assertNull(coordinator.retrieveSegmentForId(theId.getDataSource(), theId.asSegmentId().toString()));
+ Assert.assertNull(coordinator.retrieveSegmentForId(theId.asSegmentId()));
lockbox.unlock(task1, Intervals.ETERNITY);
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
index 84efcfda0bc..fac3feff14c 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
@@ -33,6 +33,7 @@ import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.client.coordinator.CoordinatorClient;
import org.apache.druid.client.coordinator.NoopCoordinatorClient;
import org.apache.druid.client.indexing.TaskStatusResponse;
+import org.apache.druid.common.utils.IdUtils;
import org.apache.druid.data.input.InputFormat;
import org.apache.druid.data.input.MaxSizeSplitHintSpec;
import org.apache.druid.data.input.impl.CSVParseSpec;
@@ -105,6 +106,7 @@ import org.apache.druid.server.security.AuthConfig;
import org.apache.druid.server.security.AuthTestUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
import org.apache.druid.utils.CompressionUtils;
import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
import org.joda.time.DateTime;
@@ -1024,10 +1026,11 @@ public class AbstractParallelIndexSupervisorTaskTest extends IngestionTestBase
public ListenableFuture<DataSegment> fetchSegment(String dataSource, String segmentId, boolean includeUnused)
{
try {
+ final SegmentId parsedSegmentId = IdUtils.getValidSegmentId(dataSource, segmentId);
final DataSegment segment = exec.submit(
() -> includeUnused
- ? getStorageCoordinator().retrieveSegmentForId(dataSource, segmentId)
- : getStorageCoordinator().retrieveUsedSegmentForId(dataSource, segmentId)
+ ? getStorageCoordinator().retrieveSegmentForId(parsedSegmentId)
+ : getStorageCoordinator().retrieveUsedSegmentForId(parsedSegmentId)
).get();
if (segment == null) {
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java b/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
index b67bd48856e..bd8d6481ba6 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/test/TestIndexerMetadataStorageCoordinator.java
@@ -221,7 +221,7 @@ public class TestIndexerMetadataStorageCoordinator implements IndexerMetadataSto
Map<DataSegment, ReplaceTaskLock> appendSegmentToReplaceLock,
DataSourceMetadata startMetadata,
DataSourceMetadata endMetadata,
- String taskGroup,
+ String taskAllocatorId,
SegmentSchemaMapping segmentSchemaMapping
)
{
@@ -299,13 +299,13 @@ public class TestIndexerMetadataStorageCoordinator implements IndexerMetadataSto
}
@Override
- public DataSegment retrieveSegmentForId(String dataSource, String segmentId)
+ public DataSegment retrieveSegmentForId(SegmentId segmentId)
{
return null;
}
@Override
- public DataSegment retrieveUsedSegmentForId(String dataSource, String segmentId)
+ public DataSegment retrieveUsedSegmentForId(SegmentId segmentId)
{
return null;
}
diff --git a/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java b/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
index f84b9350ead..db62686cd59 100644
--- a/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
+++ b/server/src/main/java/org/apache/druid/indexing/overlord/IndexerMetadataStorageCoordinator.java
@@ -358,7 +358,7 @@ public interface IndexerMetadataStorageCoordinator
Map<DataSegment, ReplaceTaskLock> appendSegmentToReplaceLock,
DataSourceMetadata startMetadata,
DataSourceMetadata endMetadata,
- String taskGroup,
+ String taskAllocatorId,
@Nullable SegmentSchemaMapping segmentSchemaMapping
);
@@ -466,9 +466,9 @@ public interface IndexerMetadataStorageCoordinator
* This option exists mainly to provide a consistent view of the metadata, for example, in calls from MSQ controller
* and worker and would generally not be required.
*/
- DataSegment retrieveSegmentForId(String dataSource, String segmentId);
+ DataSegment retrieveSegmentForId(SegmentId segmentId);
- DataSegment retrieveUsedSegmentForId(String dataSource, String segmentId);
+ DataSegment retrieveUsedSegmentForId(SegmentId segmentId);
/**
* Delete entries from the upgrade segments table after the corresponding replace task has ended
diff --git a/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java b/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
index e4f84dab233..6ef71cb83de 100644
--- a/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
+++ b/server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java
@@ -202,7 +202,7 @@ public class IndexerSQLMetadataStorageCoordinator implements IndexerMetadataStor
@Override
public List<Pair<DataSegment, String>> retrieveUsedSegmentsAndCreatedDates(String dataSource, List<Interval> intervals)
{
- return inReadWriteDatasourceTransaction(
+ return inReadOnlyDatasourceTransaction(
dataSource,
transaction -> transaction.findUsedSegmentsPlusOverlappingAnyOf(intervals)
.stream()
@@ -836,7 +836,7 @@ public class IndexerSQLMetadataStorageCoordinator implements IndexerMetadataStor
boolean reduceMetadataIO
)
{
- return inReadWriteDatasourceTransaction(
+ return inReadOnlyDatasourceTransaction(
dataSource,
transaction -> {
if (reduceMetadataIO) {
@@ -2381,7 +2381,7 @@ public class IndexerSQLMetadataStorageCoordinator implements IndexerMetadataStor
@VisibleForTesting
Set<DataSegment> retrieveUsedSegmentsForAllocation(
- final SegmentMetadataTransaction transaction,
+ final SegmentMetadataReadTransaction transaction,
final String dataSource,
final Interval interval
)
@@ -2441,22 +2441,20 @@ public class IndexerSQLMetadataStorageCoordinator implements IndexerMetadataStor
}
@Override
- public DataSegment retrieveSegmentForId(final String dataSource, final String segmentId)
+ public DataSegment retrieveSegmentForId(SegmentId segmentId)
{
- final SegmentId parsedSegmentId = IdUtils.getValidSegmentId(dataSource, segmentId);
- return inReadWriteDatasourceTransaction(
- dataSource,
- transaction -> transaction.findSegment(parsedSegmentId)
+ return inReadOnlyDatasourceTransaction(
+ segmentId.getDataSource(),
+ transaction -> transaction.findSegment(segmentId)
);
}
@Override
- public DataSegment retrieveUsedSegmentForId(String dataSource, String segmentId)
+ public DataSegment retrieveUsedSegmentForId(SegmentId segmentId)
{
- final SegmentId parsedSegmentId = IdUtils.getValidSegmentId(dataSource, segmentId);
- return inReadWriteDatasourceTransaction(
- dataSource,
- transaction -> transaction.findUsedSegment(parsedSegmentId)
+ return inReadOnlyDatasourceTransaction(
+ segmentId.getDataSource(),
+ transaction -> transaction.findUsedSegment(segmentId)
);
}
diff --git a/server/src/main/java/org/apache/druid/server/http/MetadataResource.java b/server/src/main/java/org/apache/druid/server/http/MetadataResource.java
index c2c747df4f6..5660aed929c 100644
--- a/server/src/main/java/org/apache/druid/server/http/MetadataResource.java
+++ b/server/src/main/java/org/apache/druid/server/http/MetadataResource.java
@@ -33,6 +33,7 @@ import org.apache.druid.error.InvalidInput;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
import org.apache.druid.indexing.overlord.Segments;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.metadata.SegmentsMetadataManager;
import org.apache.druid.metadata.SortOrder;
@@ -407,17 +408,26 @@ public class MetadataResource
@ResourceFilters(DatasourceResourceFilter.class)
public Response getSegment(
@PathParam("dataSourceName") String dataSourceName,
- @PathParam("segmentId") String segmentId,
+ @PathParam("segmentId") String serializedSegmentId,
@QueryParam("includeUnused") @Nullable Boolean includeUnused
)
{
+ final SegmentId segmentId = SegmentId.tryParse(dataSourceName, serializedSegmentId);
+ if (segmentId == null) {
+ return Response.status(Response.Status.BAD_REQUEST).entity(
+ StringUtils.format(
+ "Could not parse Segment ID[%s] for DataSource[%s]",
+ StringUtils.escapeHtml(serializedSegmentId),
+ StringUtils.escapeHtml(dataSourceName)
+ )
+ ).build();
+ }
+
ImmutableDruidDataSource dataSource = getDataSource(dataSourceName);
if (dataSource == null) {
return Response.status(Response.Status.NOT_FOUND).build();
- }
-
- for (SegmentId possibleSegmentId : SegmentId.iteratePossibleParsingsWithDataSource(dataSourceName, segmentId)) {
- DataSegment segment = dataSource.getSegment(possibleSegmentId);
+ } else {
+ DataSegment segment = dataSource.getSegment(segmentId);
if (segment != null) {
return Response.status(Response.Status.OK).entity(segment).build();
}
@@ -425,9 +435,9 @@ public class MetadataResource
// fallback to db
final DataSegment segment;
if (Boolean.TRUE.equals(includeUnused)) {
- segment = metadataStorageCoordinator.retrieveSegmentForId(dataSourceName, segmentId);
+ segment = metadataStorageCoordinator.retrieveSegmentForId(segmentId);
} else {
- segment = metadataStorageCoordinator.retrieveUsedSegmentForId(dataSourceName, segmentId);
+ segment = metadataStorageCoordinator.retrieveUsedSegmentForId(segmentId);
}
if (segment != null) {
diff --git a/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorReadOnlyTest.java b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorReadOnlyTest.java
new file mode 100644
index 00000000000..5f5d0db9f58
--- /dev/null
+++ b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorReadOnlyTest.java
@@ -0,0 +1,320 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.metadata;
+
+import org.apache.druid.discovery.NodeRole;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.DruidExceptionMatcher;
+import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
+import org.apache.druid.indexing.overlord.ObjectMetadata;
+import org.apache.druid.indexing.overlord.SegmentCreateRequest;
+import org.apache.druid.indexing.overlord.Segments;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.apache.druid.metadata.segment.SegmentMetadataTransactionFactory;
+import org.apache.druid.metadata.segment.SqlSegmentMetadataTransactionFactory;
+import org.apache.druid.metadata.segment.cache.HeapMemorySegmentMetadataCache;
+import org.apache.druid.metadata.segment.cache.SegmentMetadataCache;
+import org.apache.druid.segment.TestDataSource;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig;
+import org.apache.druid.server.coordinator.simulate.BlockingExecutorService;
+import org.apache.druid.server.coordinator.simulate.TestDruidLeaderSelector;
+import org.apache.druid.server.coordinator.simulate.WrappingScheduledExecutorService;
+import org.apache.druid.timeline.partition.NumberedPartialShardSpec;
+import org.hamcrest.MatcherAssert;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.function.ThrowingRunnable;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Unit tests to verify behaviour of {@link IndexerSQLMetadataStorageCoordinator}
+ * on the Coordinator for read-only purposes.
+ */
+@RunWith(Parameterized.class)
+public class IndexerSQLMetadataStorageCoordinatorReadOnlyTest extends IndexerSqlMetadataStorageCoordinatorTestBase
+{
+ @Rule
+ public final TestDerbyConnector.DerbyConnectorRule derbyConnectorRule
+ = new TestDerbyConnector.DerbyConnectorRule();
+
+ private IndexerMetadataStorageCoordinator readOnlyStorage;
+ private IndexerMetadataStorageCoordinator readWriteStorage;
+
+ private TestDruidLeaderSelector leaderSelector;
+ private SegmentMetadataCache segmentMetadataCache;
+ private StubServiceEmitter emitter;
+ private BlockingExecutorService cachePollExecutor;
+
+ private final SegmentMetadataCache.UsageMode cacheMode;
+
+ @Parameterized.Parameters(name = "cacheMode = {0}")
+ public static Object[][] testParameters()
+ {
+ return new Object[][]{
+ {SegmentMetadataCache.UsageMode.ALWAYS},
+ {SegmentMetadataCache.UsageMode.NEVER},
+ {SegmentMetadataCache.UsageMode.IF_SYNCED}
+ };
+ }
+
+ public IndexerSQLMetadataStorageCoordinatorReadOnlyTest(SegmentMetadataCache.UsageMode cacheMode)
+ {
+ this.cacheMode = cacheMode;
+ }
+
+ @Before
+ public void setup()
+ {
+ derbyConnector = derbyConnectorRule.getConnector();
+
+ leaderSelector = new TestDruidLeaderSelector();
+ emitter = new StubServiceEmitter();
+ cachePollExecutor = new BlockingExecutorService("test-cache-poll-exec");
+ segmentMetadataCache = new HeapMemorySegmentMetadataCache(
+ mapper,
+ () -> new SegmentsMetadataManagerConfig(null, cacheMode),
+ derbyConnectorRule.metadataTablesConfigSupplier(),
+ derbyConnector,
+ (corePoolSize, nameFormat) -> new WrappingScheduledExecutorService(
+ nameFormat,
+ cachePollExecutor,
+ false
+ ),
+ emitter
+ );
+
+ readOnlyStorage = createStorageCoordinator(NodeRole.COORDINATOR);
+ readWriteStorage = createStorageCoordinator(NodeRole.OVERLORD);
+
+ derbyConnector.createSegmentTable();
+ derbyConnector.createPendingSegmentsTable();
+
+ leaderSelector.becomeLeader();
+
+ // Get the cache ready if required
+ if (isCacheEnabled()) {
+ segmentMetadataCache.start();
+ segmentMetadataCache.becomeLeader();
+ syncCache();
+ syncCache();
+ }
+ }
+
+ @After
+ public void tearDown()
+ {
+ segmentMetadataCache.stopBeingLeader();
+ segmentMetadataCache.stop();
+ leaderSelector.stopBeingLeader();
+ }
+
+ private void syncCache()
+ {
+ if (isCacheEnabled()) {
+ cachePollExecutor.finishNextPendingTasks(2);
+ }
+ }
+
+ private boolean isCacheEnabled()
+ {
+ return cacheMode != SegmentMetadataCache.UsageMode.NEVER;
+ }
+
+ private IndexerSQLMetadataStorageCoordinator createStorageCoordinator(
+ NodeRole nodeRole
+ )
+ {
+ final SegmentMetadataTransactionFactory transactionFactory = new SqlSegmentMetadataTransactionFactory(
+ mapper,
+ derbyConnectorRule.metadataTablesConfigSupplier().get(),
+ derbyConnector,
+ leaderSelector,
+ Set.of(nodeRole),
+ segmentMetadataCache,
+ emitter
+ );
+
+ return new IndexerSQLMetadataStorageCoordinator(
+ transactionFactory,
+ TestHelper.JSON_MAPPER,
+ derbyConnectorRule.metadataTablesConfigSupplier().get(),
+ derbyConnector,
+ null,
+ CentralizedDatasourceSchemaConfig.enabled(false)
+ );
+ }
+
+ @Test
+ public void test_markSegmentsAsUnused_throwsException()
+ {
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.markSegmentAsUnused(defaultSegment.getId())
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.markAllSegmentsAsUnused(TestDataSource.WIKI)
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.markSegmentsAsUnused(TestDataSource.WIKI, Set.of(defaultSegment.getId()))
+ );
+ }
+
+ @Test
+ public void test_commitSegments_throwsException()
+ {
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.commitSegments(Set.of(defaultSegment), null)
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.commitSegmentsAndMetadata(Set.of(defaultSegment), null, null, null)
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.commitAppendSegments(
+ Set.of(defaultSegment),
+ Map.of(),
+ "allocator",
+ null
+ )
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.commitAppendSegmentsAndMetadata(
+ Set.of(defaultSegment),
+ Map.of(),
+ null,
+ null,
+ "allocator",
+ null
+ )
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.commitReplaceSegments(Set.of(defaultSegment), Set.of(), null)
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.commitMetadataOnly(
+ TestDataSource.WIKI,
+ new ObjectMetadata("A"),
+ new ObjectMetadata("B")
+ )
+ );
+ }
+
+ @Test
+ public void test_deleteSegments_throwsException()
+ {
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.deleteSegments(Set.of(defaultSegment))
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.deletePendingSegments(TestDataSource.WIKI)
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.deletePendingSegmentsCreatedInInterval(TestDataSource.WIKI, Intervals.ETERNITY)
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.deletePendingSegmentsForTaskAllocatorId(TestDataSource.WIKI, "allocator")
+ );
+ }
+
+ @Test
+ public void test_allocatePendingSegment_throwsException()
+ {
+ final SegmentCreateRequest createRequest =
+ new SegmentCreateRequest("seq1", null, "v1", NumberedPartialShardSpec.instance(), "allocator1");
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.allocatePendingSegment(
+ TestDataSource.WIKI,
+ Intervals.ETERNITY,
+ true,
+ createRequest
+ )
+ );
+ verifyThrowsDefensiveException(
+ () -> readOnlyStorage.allocatePendingSegments(
+ TestDataSource.WIKI,
+ Intervals.ETERNITY,
+ false,
+ List.of(createRequest),
+ true
+ )
+ );
+ }
+
+ @Test
+ public void test_retrieveSegmentForId_returnsSegment_ifPresent()
+ {
+ Assert.assertNull(
+ readOnlyStorage.retrieveSegmentForId(defaultSegment.getId())
+ );
+
+ readWriteStorage.commitSegments(Set.of(defaultSegment), null);
+ Assert.assertEquals(
+ defaultSegment,
+ readOnlyStorage.retrieveSegmentForId(defaultSegment.getId())
+ );
+ }
+
+ @Test
+ public void test_retrieveUsedSegmentForId_returnsSegment_ifPresent()
+ {
+ Assert.assertNull(
+ readOnlyStorage.retrieveUsedSegmentForId(defaultSegment.getId())
+ );
+
+ readWriteStorage.commitSegments(Set.of(defaultSegment), null);
+ Assert.assertEquals(
+ defaultSegment,
+ readOnlyStorage.retrieveUsedSegmentForId(defaultSegment.getId())
+ );
+ }
+
+ @Test
+ public void test_retrieveAllUsedSegments_returnsSegments_ifPresent()
+ {
+ Assert.assertEquals(
+ Set.of(),
+ readOnlyStorage.retrieveAllUsedSegments(defaultSegment.getDataSource(), Segments.INCLUDING_OVERSHADOWED)
+ );
+
+ readWriteStorage.commitSegments(Set.of(defaultSegment), null);
+ Assert.assertEquals(
+ Set.of(defaultSegment),
+ readOnlyStorage.retrieveAllUsedSegments(defaultSegment.getDataSource(), Segments.INCLUDING_OVERSHADOWED)
+ );
+ }
+
+ private static void verifyThrowsDefensiveException(ThrowingRunnable runnable)
+ {
+ MatcherAssert.assertThat(
+ Assert.assertThrows(DruidException.class, runnable),
+ DruidExceptionMatcher.defensive().expectMessageIs(
+ "Only Overlord can perform write transactions on segment metadata."
+ )
+ );
+ }
+}
diff --git a/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
index f7c5af17ecf..d5572c5b058 100644
--- a/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
+++ b/server/src/test/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinatorTest.java
@@ -989,7 +989,7 @@ public class IndexerSQLMetadataStorageCoordinatorTest extends IndexerSqlMetadata
coordinator.commitSegments(Set.of(defaultSegment), null);
Assert.assertEquals(
defaultSegment,
- coordinator.retrieveUsedSegmentForId(defaultSegment.getDataSource(), defaultSegment.getId().toString())
+ coordinator.retrieveUsedSegmentForId(defaultSegment.getId())
);
}
@@ -1000,7 +1000,7 @@ public class IndexerSQLMetadataStorageCoordinatorTest extends IndexerSqlMetadata
markAllSegmentsUnused(ImmutableSet.of(defaultSegment), DateTimes.nowUtc());
Assert.assertEquals(
defaultSegment,
- coordinator.retrieveSegmentForId(defaultSegment.getDataSource(), defaultSegment.getId().toString())
+ coordinator.retrieveSegmentForId(defaultSegment.getId())
);
}
@@ -2838,7 +2838,6 @@ public class IndexerSQLMetadataStorageCoordinatorTest extends IndexerSqlMetadata
}
-
@Test
public void testAllocatePendingSegmentsSkipSegmentPayloadFetch()
{
@@ -3705,7 +3704,7 @@ public class IndexerSQLMetadataStorageCoordinatorTest extends IndexerSqlMetadata
false,
"taskAllocatorId"
);
- Assert.assertNull(coordinator.retrieveSegmentForId(theId.getDataSource(), theId.asSegmentId().toString()));
+ Assert.assertNull(coordinator.retrieveSegmentForId(theId.asSegmentId()));
}
@Test
@@ -3969,7 +3968,7 @@ public class IndexerSQLMetadataStorageCoordinatorTest extends IndexerSqlMetadata
}
}
- Set<SegmentIdWithShardSpec> observed = transactionFactory.inReadWriteDatasourceTransaction(
+ Set<SegmentIdWithShardSpec> observed = transactionFactory.inReadOnlyDatasourceTransaction(
datasource,
transaction ->
coordinator.retrieveUsedSegmentsForAllocation(transaction, datasource, month)
diff --git a/server/src/test/java/org/apache/druid/server/http/MetadataResourceTest.java b/server/src/test/java/org/apache/druid/server/http/MetadataResourceTest.java
index 36452095c7c..4458fd44f6c 100644
--- a/server/src/test/java/org/apache/druid/server/http/MetadataResourceTest.java
+++ b/server/src/test/java/org/apache/druid/server/http/MetadataResourceTest.java
@@ -122,13 +122,13 @@ public class MetadataResourceTest
storageCoordinator = Mockito.mock(IndexerMetadataStorageCoordinator.class);
Mockito.doReturn(segments[4])
.when(storageCoordinator)
- .retrieveUsedSegmentForId(DATASOURCE1, segments[4].getId().toString());
+ .retrieveUsedSegmentForId(segments[4].getId());
Mockito.doReturn(null)
.when(storageCoordinator)
- .retrieveUsedSegmentForId(DATASOURCE1, segments[5].getId().toString());
+ .retrieveUsedSegmentForId(segments[5].getId());
Mockito.doReturn(segments[5])
.when(storageCoordinator)
- .retrieveSegmentForId(DATASOURCE1, segments[5].getId().toString());
+ .retrieveSegmentForId(segments[5].getId());
Mockito.doAnswer(mockIterateAllUnusedSegmentsForDatasource())
.when(storageCoordinator)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org