Skip to content

Fix the broken Appenderator contract in KafkaIndexTask #5905

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
gianm merged 4 commits into from
Jul 3, 2018
Merged

Fix the broken Appenderator contract in KafkaIndexTask #5905

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7aabe77
Fix broken Appenderator contract in KafkaIndexTask
jihoonson Jun 22, 2018
f719d68
fix build
jihoonson Jun 26, 2018
32911a7
add publishFuture
jihoonson Jun 26, 2018
f337d62
reuse sequenceToUse if possible
jihoonson Jun 27, 2018
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
508 changes: 261 additions & 247 deletions ...vice/src/main/java/io/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java
View file
Open in desktop

Large diffs are not rendered by default.

11 changes: 0 additions & 11 deletions ...ions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIOConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,15 +31,13 @@
public class KafkaIOConfig implements IOConfig
{
private static final boolean DEFAULT_USE_TRANSACTION = true;
private static final boolean DEFAULT_PAUSE_AFTER_READ = false;
private static final boolean DEFAULT_SKIP_OFFSET_GAPS = false;

private final String baseSequenceName;
private final KafkaPartitions startPartitions;
private final KafkaPartitions endPartitions;
private final Map<String, String> consumerProperties;
private final boolean useTransaction;
private final boolean pauseAfterRead;
private final Optional<DateTime> minimumMessageTime;
private final Optional<DateTime> maximumMessageTime;
private final boolean skipOffsetGaps;
Expand All @@ -51,7 +49,6 @@ public KafkaIOConfig(
@JsonProperty("endPartitions") KafkaPartitions endPartitions,
@JsonProperty("consumerProperties") Map<String, String> consumerProperties,
@JsonProperty("useTransaction") Boolean useTransaction,
@JsonProperty("pauseAfterRead") Boolean pauseAfterRead,
@JsonProperty("minimumMessageTime") DateTime minimumMessageTime,
@JsonProperty("maximumMessageTime") DateTime maximumMessageTime,
@JsonProperty("skipOffsetGaps") Boolean skipOffsetGaps
Expand All @@ -62,7 +59,6 @@ public KafkaIOConfig(
this.endPartitions = Preconditions.checkNotNull(endPartitions, "endPartitions");
this.consumerProperties = Preconditions.checkNotNull(consumerProperties, "consumerProperties");
this.useTransaction = useTransaction != null ? useTransaction : DEFAULT_USE_TRANSACTION;
this.pauseAfterRead = pauseAfterRead != null ? pauseAfterRead : DEFAULT_PAUSE_AFTER_READ;
this.minimumMessageTime = Optional.fromNullable(minimumMessageTime);
this.maximumMessageTime = Optional.fromNullable(maximumMessageTime);
this.skipOffsetGaps = skipOffsetGaps != null ? skipOffsetGaps : DEFAULT_SKIP_OFFSET_GAPS;
Expand Down Expand Up @@ -117,12 +113,6 @@ public boolean isUseTransaction()
return useTransaction;
}

@JsonProperty
public boolean isPauseAfterRead()
{
return pauseAfterRead;
}

@JsonProperty
public Optional<DateTime> getMaximumMessageTime()
{
Expand Down Expand Up @@ -150,7 +140,6 @@ public String toString()
", endPartitions=" + endPartitions +
", consumerProperties=" + consumerProperties +
", useTransaction=" + useTransaction +
", pauseAfterRead=" + pauseAfterRead +
", minimumMessageTime=" + minimumMessageTime +
", maximumMessageTime=" + maximumMessageTime +
", skipOffsetGaps=" + skipOffsetGaps +
Expand Down
6 changes: 0 additions & 6 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 @@ -68,8 +68,6 @@

public class KafkaIndexTask extends AbstractTask implements ChatHandler
{
public static final long PAUSE_FOREVER = -1L;

public enum Status
{
NOT_STARTED,
Expand All @@ -86,14 +84,11 @@ public enum Status
private static final Random RANDOM = new Random();
static final long POLL_TIMEOUT = 100;
static final long LOCK_ACQUIRE_TIMEOUT_SECONDS = 15;
private static final String METADATA_NEXT_PARTITIONS = "nextPartitions";
private static final String METADATA_PUBLISH_PARTITIONS = "publishPartitions";

private final DataSchema dataSchema;
private final InputRowParser<ByteBuffer> parser;
private final KafkaTuningConfig tuningConfig;
private final KafkaIOConfig ioConfig;
private final AuthorizerMapper authorizerMapper;
private final Optional<ChatHandlerProvider> chatHandlerProvider;
private final KafkaIndexTaskRunner runner;

Expand Down Expand Up @@ -126,7 +121,6 @@ public KafkaIndexTask(
this.tuningConfig = Preconditions.checkNotNull(tuningConfig, "tuningConfig");
this.ioConfig = Preconditions.checkNotNull(ioConfig, "ioConfig");
this.chatHandlerProvider = Optional.fromNullable(chatHandlerProvider);
this.authorizerMapper = authorizerMapper;
final CircularBuffer<Throwable> savedParseExceptions;
if (tuningConfig.getMaxSavedParseExceptions() > 0) {
savedParseExceptions = new CircularBuffer<>(tuningConfig.getMaxSavedParseExceptions());
Expand Down
27 changes: 8 additions & 19 deletions ...re/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,11 +28,11 @@
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import io.druid.indexer.TaskLocation;
import io.druid.indexer.TaskStatus;
import io.druid.indexing.common.RetryPolicy;
import io.druid.indexing.common.RetryPolicyConfig;
import io.druid.indexing.common.RetryPolicyFactory;
import io.druid.indexing.common.TaskInfoProvider;
import io.druid.indexer.TaskStatus;
import io.druid.java.util.common.IAE;
import io.druid.java.util.common.IOE;
import io.druid.java.util.common.ISE;
Expand Down Expand Up @@ -167,19 +167,14 @@ public boolean resume(final String id)

public Map<Integer, Long> pause(final String id)
{
return pause(id, 0);
}

public Map<Integer, Long> pause(final String id, final long timeout)
{
log.debug("Pause task[%s] timeout[%d]", id, timeout);
log.debug("Pause task[%s]", id);

try {
final FullResponseHolder response = submitRequest(
id,
HttpMethod.POST,
"pause",
timeout > 0 ? StringUtils.format("timeout=%d", timeout) : null,
null,
true
);

Expand Down Expand Up @@ -361,18 +356,17 @@ public Map<Integer, Long> getEndOffsets(final String id)
public boolean setEndOffsets(
final String id,
final Map<Integer, Long> endOffsets,
final boolean resume,
final boolean finalize
)
{
log.debug("SetEndOffsets task[%s] endOffsets[%s] resume[%s] finalize[%s]", id, endOffsets, resume, finalize);
log.debug("SetEndOffsets task[%s] endOffsets[%s] finalize[%s]", id, endOffsets, finalize);

try {
final FullResponseHolder response = submitRequest(
id,
HttpMethod.POST,
"offsets/end",
StringUtils.format("resume=%s&finish=%s", resume, finalize),
StringUtils.format("finish=%s", finalize),
jsonMapper.writeValueAsBytes(endOffsets),
true
);
Expand Down Expand Up @@ -415,19 +409,14 @@ public Boolean call()
}

public ListenableFuture<Map<Integer, Long>> pauseAsync(final String id)
{
return pauseAsync(id, 0);
}

public ListenableFuture<Map<Integer, Long>> pauseAsync(final String id, final long timeout)
{
return executorService.submit(
new Callable<Map<Integer, Long>>()
{
@Override
public Map<Integer, Long> call()
{
return pause(id, timeout);
return pause(id);
}
}
);
Expand Down Expand Up @@ -490,7 +479,7 @@ public Map<Integer, Long> call()
}

public ListenableFuture<Boolean> setEndOffsetsAsync(
final String id, final Map<Integer, Long> endOffsets, final boolean resume, final boolean finalize
final String id, final Map<Integer, Long> endOffsets, final boolean finalize
)
{
return executorService.submit(
Expand All @@ -499,7 +488,7 @@ public ListenableFuture<Boolean> setEndOffsetsAsync(
@Override
public Boolean call()
{
return setEndOffsets(id, endOffsets, resume, finalize);
return setEndOffsets(id, endOffsets, finalize);
}
}
);
Expand Down
3 changes: 1 addition & 2 deletions ...re/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTaskRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,12 +60,11 @@ public interface KafkaIndexTaskRunner extends ChatHandler
@VisibleForTesting
Response setEndOffsets(
Map<Integer, Long> offsets,
boolean resume,
boolean finish // this field is only for internal purposes, shouldn't be usually set by users
) throws InterruptedException;

@VisibleForTesting
Response pause(long timeout) throws InterruptedException;
Response pause() throws InterruptedException;
@VisibleForTesting
void resume() throws InterruptedException;
}
Loading