Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions docs/configuration/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -1414,6 +1414,7 @@ Additional Peon configs include:
|`druid.indexer.task.ignoreTimestampSpecForDruidInputSource`|If true, tasks using the [Druid input source](../ingestion/input-sources.md) will ignore the provided timestampSpec, and will use the `__time` column of the input datasource. This option is provided for compatibility with ingestion specs written before Druid 0.22.0.|false|
|`druid.indexer.task.storeEmptyColumns`|Boolean value for whether or not to store empty columns during ingestion. When set to true, Druid stores every column specified in the [`dimensionsSpec`](../ingestion/ingestion-spec.md#dimensionsspec). If you use the string-based schemaless ingestion and don't specify any dimensions to ingest, you must also set [`includeAllDimensions`](../ingestion/ingestion-spec.md#dimensionsspec) for Druid to store empty columns.<br/><br/>If you set `storeEmptyColumns` to false, Druid SQL queries referencing empty columns will fail. If you intend to leave `storeEmptyColumns` disabled, you should either ingest placeholder data for empty columns or else not query on empty columns.<br/><br/>You can overwrite this configuration by setting `storeEmptyColumns` in the [task context](../ingestion/tasks.md#context-parameters).|true|
|`druid.indexer.task.tmpStorageBytesPerTask`|Maximum number of bytes per task to be used to store temporary files on disk. This config is generally intended for internal usage. Attempts to set it are very likely to be overwritten by the TaskRunner that executes the task, so be sure of what you expect to happen before directly adjusting this configuration parameter. The config is documented here primarily to provide an understanding of what it means if/when someone sees that it has been set. A value of -1 disables this limit. |-1|
|`druid.indexer.task.allowHadoopTaskExecution`|Conditional dictating if the cluster allows `index_hadoop` tasks to be executed. `index_hadoop` is deprecated, and defaulting to false will force cluster operators to acknowledge the deprecation and consciously opt in to using index_hadoop with the understanding that it will be removed in the future.|false|
|`druid.indexer.server.maxChatRequests`|Maximum number of concurrent requests served by a task's chat handler. Set to 0 to disable limiting.|0|

If the Peon is running in remote mode, there must be an Overlord up and running. Peons in remote mode can set the following configurations:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -295,6 +295,22 @@ public TaskStatus runTask(TaskToolbox toolbox)
{
try {
taskConfig = toolbox.getConfig();
if (!taskConfig.isAllowHadoopTaskExecution()) {
errorMsg = StringUtils.format(
"Hadoop tasks are deprecated and will be removed in a future release. "
+ "Currently, they are not allowed to run on this cluster. If you wish to run them despite deprecation, "
+ "please set [%s] to true.",
TaskConfig.ALLOW_HADOOP_TASK_EXECUTION_KEY
);
log.error(errorMsg);
toolbox.getTaskReportFileWriter().write(getId(), getTaskCompletionReports());
return TaskStatus.failure(getId(), errorMsg);
}
log.warn("Running deprecated index_hadoop task [%s]. "
+ "Hadoop batch indexing is deprecated and will be removed in a future release. "
+ "Please plan your migration to one of Druid's supported indexing patterns.",
getId()
);
if (chatHandlerProvider.isPresent()) {
log.info("Found chat handler of class[%s]", chatHandlerProvider.get().getClass().getName());
chatHandlerProvider.get().register(getId(), this, false);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,9 @@
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.apache.druid.indexer.granularity.UniformGranularitySpec;
import org.apache.druid.indexing.common.TaskToolbox;
import org.apache.druid.indexing.common.config.TaskConfigBuilder;
import org.apache.druid.indexing.overlord.TestTaskToolboxFactory;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.granularity.Granularities;
Expand All @@ -41,6 +44,45 @@
{
private final ObjectMapper jsonMapper = new DefaultObjectMapper();

@Test
public void testHadoopTaskWontRunWithDefaultTaskConfig()
{
final HadoopIndexTask task = new HadoopIndexTask(
null,
new HadoopIngestionSpec(
DataSchema.builder()
.withDataSource("foo")
.withGranularity(
new UniformGranularitySpec(
Granularities.DAY,
null,
ImmutableList.of(Intervals.of("2010-01-01/P1D"))
Comment on lines +50 to +59

Check notice

Code scanning / CodeQL

Deprecated method or constructor invocation Note test

Invoking
Builder.withObjectMapper
should be avoided because it has been deprecated.
)
)
.withObjectMapper(jsonMapper)
.build(),
new HadoopIOConfig(ImmutableMap.of("paths", "bar"), null, null),
null
),
null,
null,
"blah",
jsonMapper,
null,
AuthTestUtils.TEST_AUTHORIZER_MAPPER,
null,
new HadoopTaskConfig(null, null)
);

TestTaskToolboxFactory.Builder builder = new TestTaskToolboxFactory.Builder().setConfig(new TaskConfigBuilder().build());
TaskToolbox toolbox = new TestTaskToolboxFactory(builder).build(task);

Assert.assertEquals("Hadoop tasks are deprecated and will be removed in a future release. Currently, "
+ "they are not allowed to run on this cluster. If you wish to run them despite deprecation, "
+ "please set [druid.indexer.task.allowHadoopTaskExecution] to true.",
task.runTask(toolbox).getErrorMsg());
}

@Test
public void testCorrectInputSourceResources()
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
*/
public class TaskConfig implements TaskDirectory
{
public static final String ALLOW_HADOOP_TASK_EXECUTION_KEY = "druid.indexer.task.allowHadoopTaskExecution";
private static final Period DEFAULT_DIRECTORY_LOCK_TIMEOUT = new Period("PT10M");
private static final Period DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT = new Period("PT5M");
private static final boolean DEFAULT_STORE_EMPTY_COLUMNS = true;
Expand Down Expand Up @@ -76,6 +77,9 @@ public class TaskConfig implements TaskDirectory
@JsonProperty
private final long tmpStorageBytesPerTask;

@JsonProperty
private final boolean allowHadoopTaskExecution;

@JsonCreator
public TaskConfig(
@JsonProperty("baseDir") String baseDir,
Expand All @@ -87,7 +91,8 @@ public TaskConfig(
@JsonProperty("ignoreTimestampSpecForDruidInputSource") boolean ignoreTimestampSpecForDruidInputSource,
@JsonProperty("storeEmptyColumns") @Nullable Boolean storeEmptyColumns,
@JsonProperty("encapsulatedTask") boolean enableTaskLevelLogPush,
@JsonProperty("tmpStorageBytesPerTask") @Nullable Long tmpStorageBytesPerTask
@JsonProperty("tmpStorageBytesPerTask") @Nullable Long tmpStorageBytesPerTask,
@JsonProperty("allowHadoopTaskExecution") boolean allowHadoopTaskExecution
)
{
this.baseDir = Configs.valueOrDefault(baseDir, System.getProperty("java.io.tmpdir"));
Expand All @@ -113,6 +118,7 @@ public TaskConfig(

this.storeEmptyColumns = Configs.valueOrDefault(storeEmptyColumns, DEFAULT_STORE_EMPTY_COLUMNS);
this.tmpStorageBytesPerTask = Configs.valueOrDefault(tmpStorageBytesPerTask, DEFAULT_TMP_STORAGE_BYTES_PER_TASK);
this.allowHadoopTaskExecution = allowHadoopTaskExecution;
}

private TaskConfig(
Expand All @@ -125,7 +131,8 @@ private TaskConfig(
boolean ignoreTimestampSpecForDruidInputSource,
boolean storeEmptyColumns,
boolean encapsulatedTask,
long tmpStorageBytesPerTask
long tmpStorageBytesPerTask,
boolean allowHadoopTaskExecution
)
{
this.baseDir = baseDir;
Expand All @@ -138,6 +145,7 @@ private TaskConfig(
this.storeEmptyColumns = storeEmptyColumns;
this.encapsulatedTask = encapsulatedTask;
this.tmpStorageBytesPerTask = tmpStorageBytesPerTask;
this.allowHadoopTaskExecution = allowHadoopTaskExecution;
}

@JsonProperty
Expand Down Expand Up @@ -230,6 +238,12 @@ public long getTmpStorageBytesPerTask()
return tmpStorageBytesPerTask;
}

@JsonProperty
public boolean isAllowHadoopTaskExecution()
{
return allowHadoopTaskExecution;
}

private String defaultDir(@Nullable String configParameter, final String defaultVal)
{
if (configParameter == null) {
Expand All @@ -251,7 +265,8 @@ public TaskConfig withBaseTaskDir(File baseTaskDir)
ignoreTimestampSpecForDruidInputSource,
storeEmptyColumns,
encapsulatedTask,
tmpStorageBytesPerTask
tmpStorageBytesPerTask,
allowHadoopTaskExecution
);
}

Expand All @@ -267,7 +282,8 @@ public TaskConfig withTmpStorageBytesPerTask(long tmpStorageBytesPerTask)
ignoreTimestampSpecForDruidInputSource,
storeEmptyColumns,
encapsulatedTask,
tmpStorageBytesPerTask
tmpStorageBytesPerTask,
allowHadoopTaskExecution
);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ public class TaskConfigBuilder
private Boolean storeEmptyColumns;
private boolean enableTaskLevelLogPush;
private Long tmpStorageBytesPerTask;
private boolean allowHadoopTaskExecution;

public TaskConfigBuilder setBaseDir(String baseDir)
{
Expand Down Expand Up @@ -97,6 +98,12 @@ public TaskConfigBuilder setTmpStorageBytesPerTask(Long tmpStorageBytesPerTask)
return this;
}

public TaskConfigBuilder setAllowHadoopTaskExecution(boolean allowHadoopTaskExecution)
{
this.allowHadoopTaskExecution = allowHadoopTaskExecution;
return this;
}

public TaskConfig build()
{
return new TaskConfig(
Expand All @@ -109,7 +116,8 @@ public TaskConfig build()
ignoreTimestampSpecForDruidInputSource,
storeEmptyColumns,
enableTaskLevelLogPush,
tmpStorageBytesPerTask
tmpStorageBytesPerTask,
allowHadoopTaskExecution
);
}
}
Loading