Skip to content

internal-discovery: interfaces for announcement/discovery, curator based impls #4634

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cheddar merged 9 commits into from
Aug 16, 2017
Merged

internal-discovery: interfaces for announcement/discovery, curator based impls #4634

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
2ee2016
internal-discovery: interfaces for announcement/discovery, curator impls
himanshug Aug 1, 2017
5d8b038
more tests
himanshug Aug 7, 2017
e47b63e
address some review comments
himanshug Aug 9, 2017
013a0bc
more fixes
himanshug Aug 10, 2017
3eb23b3
address more review comments
himanshug Aug 14, 2017
f928a46
simplify ObjectMapper setup in CuratorDruidNodeAnnouncerAndDiscoveryTest
himanshug Aug 14, 2017
505ba70
fix KafkaIndexTaskTest
himanshug Aug 16, 2017
76bfb5e
make lookupTier overridable via RealtimeIndexTask and KafkaIndexTask …
himanshug Aug 16, 2017
8c46ccc
make teamcity build happy
himanshug Aug 16, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions ...ons-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,9 @@
import io.druid.data.input.Committer;
import io.druid.data.input.InputRow;
import io.druid.data.input.impl.InputRowParser;
import io.druid.discovery.DiscoveryDruidNode;
import io.druid.discovery.DruidNodeDiscoveryProvider;
import io.druid.discovery.LookupNodeService;
import io.druid.indexing.appenderator.ActionBasedSegmentAllocator;
import io.druid.indexing.appenderator.ActionBasedUsedSegmentChecker;
import io.druid.indexing.common.TaskStatus;
Expand All @@ -50,6 +53,7 @@
import io.druid.indexing.common.actions.SegmentTransactionalInsertAction;
import io.druid.indexing.common.actions.TaskActionClient;
import io.druid.indexing.common.task.AbstractTask;
import io.druid.indexing.common.task.RealtimeIndexTask;
import io.druid.indexing.common.task.TaskResource;
import io.druid.java.util.common.ISE;
import io.druid.java.util.common.StringUtils;
Expand Down Expand Up @@ -289,12 +293,25 @@ public TaskStatus run(final TaskToolbox toolbox) throws Exception
)
);

LookupNodeService lookupNodeService = getContextValue(RealtimeIndexTask.CTX_KEY_LOOKUP_TIER) == null ?
toolbox.getLookupNodeService() :
new LookupNodeService((String) getContextValue(RealtimeIndexTask.CTX_KEY_LOOKUP_TIER));
DiscoveryDruidNode discoveryDruidNode = new DiscoveryDruidNode(
toolbox.getDruidNode(),
DruidNodeDiscoveryProvider.NODE_TYPE_PEON,
ImmutableMap.of(
toolbox.getDataNodeService().getName(), toolbox.getDataNodeService(),
Copy link
Copy Markdown
Contributor

@cheddar cheddar Aug 8, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it would be nicer to take the informatino for this in the task spec instead of runtime config. That would give us the ability to set it on a per-task basis.

Copy link
Copy Markdown
Contributor Author

@himanshug himanshug Aug 16, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

lookupTier is something that user might want to override so made it overridable via task context key lookupTier in Realtime and Kafka indexing tasks.

lookupNodeService.getName(), lookupNodeService
)
);

try (
final Appenderator appenderator0 = newAppenderator(fireDepartmentMetrics, toolbox);
final AppenderatorDriver driver = newDriver(appenderator0, toolbox, fireDepartmentMetrics);
final KafkaConsumer<byte[], byte[]> consumer = newConsumer()
) {
toolbox.getDataSegmentServerAnnouncer().announce();
toolbox.getDruidNodeAnnouncer().announce(discoveryDruidNode);

appenderator = appenderator0;

Expand Down Expand Up @@ -597,6 +614,7 @@ public String apply(DataSegment input)
}
}

toolbox.getDruidNodeAnnouncer().unannounce(discoveryDruidNode);
toolbox.getDataSegmentServerAnnouncer().unannounce();

return success();
Expand Down
11 changes: 10 additions & 1 deletion ...core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaIndexTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,9 @@
import io.druid.data.input.impl.JSONPathSpec;
import io.druid.data.input.impl.StringInputRowParser;
import io.druid.data.input.impl.TimestampSpec;
import io.druid.discovery.DataNodeService;
import io.druid.discovery.DruidNodeAnnouncer;
import io.druid.discovery.LookupNodeService;
import io.druid.indexing.common.SegmentLoaderFactory;
import io.druid.indexing.common.TaskLock;
import io.druid.indexing.common.TaskStatus;
Expand Down Expand Up @@ -117,7 +120,9 @@
import io.druid.segment.loading.StorageLocationConfig;
import io.druid.segment.realtime.plumber.SegmentHandoffNotifier;
import io.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import io.druid.server.DruidNode;
import io.druid.server.coordination.DataSegmentServerAnnouncer;
import io.druid.server.coordination.ServerType;
import io.druid.timeline.DataSegment;
import org.apache.curator.test.TestingCluster;
import org.apache.kafka.clients.producer.KafkaProducer;
Expand Down Expand Up @@ -1637,7 +1642,11 @@ public List<StorageLocationConfig> getLocations()
testUtils.getTestIndexIO(),
MapCache.create(1024),
new CacheConfig(),
testUtils.getTestIndexMergerV9()
testUtils.getTestIndexMergerV9(),
EasyMock.createNiceMock(DruidNodeAnnouncer.class),
EasyMock.createNiceMock(DruidNode.class),
new LookupNodeService("tier"),
new DataNodeService("tier", 1, ServerType.INDEXER_EXECUTOR, 0)
);
}

Expand Down
39 changes: 38 additions & 1 deletion indexing-service/src/main/java/io/druid/indexing/common/TaskToolbox.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@
import com.metamx.metrics.MonitorScheduler;
import io.druid.client.cache.Cache;
import io.druid.client.cache.CacheConfig;
import io.druid.discovery.DataNodeService;
import io.druid.discovery.DruidNodeAnnouncer;
import io.druid.discovery.LookupNodeService;
import io.druid.indexing.common.actions.SegmentInsertAction;
import io.druid.indexing.common.actions.TaskActionClient;
import io.druid.indexing.common.config.TaskConfig;
Expand All @@ -44,6 +47,7 @@
import io.druid.segment.loading.SegmentLoader;
import io.druid.segment.loading.SegmentLoadingException;
import io.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import io.druid.server.DruidNode;
import io.druid.server.coordination.DataSegmentAnnouncer;
import io.druid.server.coordination.DataSegmentServerAnnouncer;
import io.druid.timeline.DataSegment;
Expand Down Expand Up @@ -87,6 +91,11 @@ public class TaskToolbox
private final CacheConfig cacheConfig;
private final IndexMergerV9 indexMergerV9;

private final DruidNodeAnnouncer druidNodeAnnouncer;
private final DruidNode druidNode;
private final LookupNodeService lookupNodeService;
private final DataNodeService dataNodeService;

public TaskToolbox(
TaskConfig config,
TaskActionClient taskActionClient,
Expand All @@ -107,7 +116,11 @@ public TaskToolbox(
IndexIO indexIO,
Cache cache,
CacheConfig cacheConfig,
IndexMergerV9 indexMergerV9
IndexMergerV9 indexMergerV9,
DruidNodeAnnouncer druidNodeAnnouncer,
DruidNode druidNode,
LookupNodeService lookupNodeService,
DataNodeService dataNodeService
)
{
this.config = config;
Expand All @@ -130,6 +143,10 @@ public TaskToolbox(
this.cache = cache;
this.cacheConfig = cacheConfig;
this.indexMergerV9 = Preconditions.checkNotNull(indexMergerV9, "Null IndexMergerV9");
this.druidNodeAnnouncer = druidNodeAnnouncer;
this.druidNode = druidNode;
this.lookupNodeService = lookupNodeService;
this.dataNodeService = dataNodeService;
}

public TaskConfig getConfig()
Expand Down Expand Up @@ -271,4 +288,24 @@ public File getPersistDir()
{
return new File(taskWorkDir, "persist");
}

public DruidNodeAnnouncer getDruidNodeAnnouncer()
{
return druidNodeAnnouncer;
}

public LookupNodeService getLookupNodeService()
{
return lookupNodeService;
}

public DataNodeService getDataNodeService()
{
return dataNodeService;
}

public DruidNode getDruidNode()
{
return druidNode;
}
}
25 changes: 23 additions & 2 deletions indexing-service/src/main/java/io/druid/indexing/common/TaskToolboxFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,11 @@
import com.metamx.metrics.MonitorScheduler;
import io.druid.client.cache.Cache;
import io.druid.client.cache.CacheConfig;
import io.druid.discovery.DataNodeService;
import io.druid.discovery.DruidNodeAnnouncer;
import io.druid.discovery.LookupNodeService;
import io.druid.guice.annotations.Processing;
import io.druid.guice.annotations.RemoteChatHandler;
import io.druid.indexing.common.actions.TaskActionClientFactory;
import io.druid.indexing.common.config.TaskConfig;
import io.druid.indexing.common.task.Task;
Expand All @@ -39,6 +43,7 @@
import io.druid.segment.loading.DataSegmentMover;
import io.druid.segment.loading.DataSegmentPusher;
import io.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import io.druid.server.DruidNode;
import io.druid.server.coordination.DataSegmentAnnouncer;
import io.druid.server.coordination.DataSegmentServerAnnouncer;

Expand Down Expand Up @@ -69,6 +74,10 @@ public class TaskToolboxFactory
private final Cache cache;
private final CacheConfig cacheConfig;
private final IndexMergerV9 indexMergerV9;
private final DruidNodeAnnouncer druidNodeAnnouncer;
private final DruidNode druidNode;
private final LookupNodeService lookupNodeService;
private final DataNodeService dataNodeService;

@Inject
public TaskToolboxFactory(
Expand All @@ -90,7 +99,11 @@ public TaskToolboxFactory(
IndexIO indexIO,
Cache cache,
CacheConfig cacheConfig,
IndexMergerV9 indexMergerV9
IndexMergerV9 indexMergerV9,
DruidNodeAnnouncer druidNodeAnnouncer,
@RemoteChatHandler DruidNode druidNode,
LookupNodeService lookupNodeService,
DataNodeService dataNodeService
)
{
this.config = config;
Expand All @@ -112,6 +125,10 @@ public TaskToolboxFactory(
this.cache = cache;
this.cacheConfig = cacheConfig;
this.indexMergerV9 = indexMergerV9;
this.druidNodeAnnouncer = druidNodeAnnouncer;
this.druidNode = druidNode;
this.lookupNodeService = lookupNodeService;
this.dataNodeService = dataNodeService;
}

public TaskToolbox build(Task task)
Expand All @@ -137,7 +154,11 @@ public TaskToolbox build(Task task)
indexIO,
cache,
cacheConfig,
indexMergerV9
indexMergerV9,
druidNodeAnnouncer,
druidNode,
lookupNodeService,
dataNodeService
);
}
}
20 changes: 20 additions & 0 deletions indexing-service/src/main/java/io/druid/indexing/common/task/RealtimeIndexTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,9 @@
import io.druid.data.input.Committer;
import io.druid.data.input.Firehose;
import io.druid.data.input.FirehoseFactory;
import io.druid.discovery.DiscoveryDruidNode;
import io.druid.discovery.DruidNodeDiscoveryProvider;
import io.druid.discovery.LookupNodeService;
import io.druid.indexing.common.TaskLock;
import io.druid.indexing.common.TaskStatus;
import io.druid.indexing.common.TaskToolbox;
Expand Down Expand Up @@ -78,6 +81,8 @@

public class RealtimeIndexTask extends AbstractTask
{
public static final String CTX_KEY_LOOKUP_TIER = "lookupTier";

private static final EmittingLogger log = new EmittingLogger(RealtimeIndexTask.class);
private final static Random random = new Random();

Expand Down Expand Up @@ -324,8 +329,22 @@ public String getVersion(final Interval interval)
Supplier<Committer> committerSupplier = null;
final File firehoseTempDir = toolbox.getFirehoseTemporaryDir();

LookupNodeService lookupNodeService = getContextValue(CTX_KEY_LOOKUP_TIER) == null ?
toolbox.getLookupNodeService() :
new LookupNodeService((String) getContextValue(CTX_KEY_LOOKUP_TIER));
DiscoveryDruidNode discoveryDruidNode = new DiscoveryDruidNode(
toolbox.getDruidNode(),
DruidNodeDiscoveryProvider.NODE_TYPE_PEON,
ImmutableMap.of(
toolbox.getDataNodeService().getName(), toolbox.getDataNodeService(),
lookupNodeService.getName(), lookupNodeService
)
);

try {
toolbox.getDataSegmentServerAnnouncer().announce();
toolbox.getDruidNodeAnnouncer().announce(discoveryDruidNode);


plumber.startJob();

Expand Down Expand Up @@ -431,6 +450,7 @@ public void run()
}

toolbox.getDataSegmentServerAnnouncer().unannounce();
toolbox.getDruidNodeAnnouncer().unannounce(discoveryDruidNode);
}

log.info("Job done!");
Expand Down
6 changes: 5 additions & 1 deletion indexing-service/src/test/java/io/druid/indexing/common/TaskToolboxTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,11 @@ public void setUp() throws IOException
mockIndexIO,
mockCache,
mockCacheConfig,
mockIndexMergerV9
mockIndexMergerV9,
null,
null,
null,
null
);
}

Expand Down
2 changes: 1 addition & 1 deletion indexing-service/src/test/java/io/druid/indexing/common/task/IndexTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -950,7 +950,7 @@ public Map<String, Object> makeLoadSpec(URI uri)
throw new UnsupportedOperationException();
}
}, null, null, null, null, null, null, null, null, null, null, jsonMapper, temporaryFolder.newFolder(),
indexIO, null, null, indexMergerV9
indexIO, null, null, indexMergerV9, null, null, null, null
)
);

Expand Down
11 changes: 10 additions & 1 deletion indexing-service/src/test/java/io/druid/indexing/common/task/RealtimeIndexTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,9 @@
import io.druid.data.input.InputRow;
import io.druid.data.input.MapBasedInputRow;
import io.druid.data.input.impl.InputRowParser;
import io.druid.discovery.DataNodeService;
import io.druid.discovery.DruidNodeAnnouncer;
import io.druid.discovery.LookupNodeService;
import io.druid.indexing.common.SegmentLoaderFactory;
import io.druid.indexing.common.TaskStatus;
import io.druid.indexing.common.TaskToolbox;
Expand Down Expand Up @@ -104,7 +107,9 @@
import io.druid.segment.realtime.plumber.SegmentHandoffNotifier;
import io.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import io.druid.segment.realtime.plumber.ServerTimeRejectionPolicyFactory;
import io.druid.server.DruidNode;
import io.druid.server.coordination.DataSegmentServerAnnouncer;
import io.druid.server.coordination.ServerType;
import io.druid.timeline.DataSegment;
import org.easymock.EasyMock;
import org.hamcrest.CoreMatchers;
Expand Down Expand Up @@ -1048,7 +1053,11 @@ public List<StorageLocationConfig> getLocations()
testUtils.getTestIndexIO(),
MapCache.create(1024),
new CacheConfig(),
testUtils.getTestIndexMergerV9()
testUtils.getTestIndexMergerV9(),
EasyMock.createNiceMock(DruidNodeAnnouncer.class),
EasyMock.createNiceMock(DruidNode.class),
new LookupNodeService("tier"),
new DataNodeService("tier", 1000, ServerType.INDEXER_EXECUTOR, 0)
);

return toolboxFactory.build(task);
Expand Down
3 changes: 2 additions & 1 deletion indexing-service/src/test/java/io/druid/indexing/common/task/SameIntervalMergeTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,8 @@ public void cleanup(DataSegment segment) throws SegmentLoadingException
{
}
}, jsonMapper, temporaryFolder.newFolder(),
indexIO, null, null, EasyMock.createMock(IndexMergerV9.class)
indexIO, null, null, EasyMock.createMock(IndexMergerV9.class),
null, null, null, null
)
);

Expand Down
6 changes: 5 additions & 1 deletion ...ng-service/src/test/java/io/druid/indexing/firehose/IngestSegmentFirehoseFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -292,7 +292,11 @@ public List<StorageLocationConfig> getLocations()
INDEX_IO,
null,
null,
INDEX_MERGER_V9
INDEX_MERGER_V9,
null,
null,
null,
null
);
Collection<Object[]> values = new LinkedList<>();
for (InputRowParser parser : Arrays.<InputRowParser>asList(
Expand Down
6 changes: 5 additions & 1 deletion ...ce/src/test/java/io/druid/indexing/firehose/IngestSegmentFirehoseFactoryTimelineTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,7 +333,11 @@ public List<StorageLocationConfig> getLocations()
INDEX_IO,
null,
null,
INDEX_MERGER_V9
INDEX_MERGER_V9,
null,
null,
null,
null
);
final Injector injector = Guice.createInjector(
new Module()
Expand Down
Loading