jihoonson commented on a change in pull request #6431: Add Kinesis Indexing Service to core Druid URL: https://github.com/apache/incubator-druid/pull/6431#discussion_r241628362
########## File path: indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClient.java ########## @@ -0,0 +1,370 @@ +/* + * 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.indexing.seekablestream; + +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableMap; +import com.google.common.util.concurrent.ListenableFuture; +import org.apache.druid.indexing.common.IndexTaskClient; +import org.apache.druid.indexing.common.TaskInfoProvider; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.RE; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.java.util.common.jackson.JacksonUtils; +import org.apache.druid.java.util.emitter.EmittingLogger; +import org.apache.druid.java.util.http.client.HttpClient; +import org.apache.druid.java.util.http.client.response.FullResponseHolder; +import org.jboss.netty.handler.codec.http.HttpMethod; +import org.jboss.netty.handler.codec.http.HttpResponseStatus; +import org.joda.time.DateTime; +import org.joda.time.Duration; + +import javax.annotation.Nullable; +import java.io.IOException; +import java.util.Collections; +import java.util.Map; +import java.util.TreeMap; + +public abstract class SeekableStreamIndexTaskClient<PartitionType, SequenceType> extends IndexTaskClient +{ + private final EmittingLogger log = new EmittingLogger(this.getClass()); + + public SeekableStreamIndexTaskClient( + HttpClient httpClient, + ObjectMapper jsonMapper, + TaskInfoProvider taskInfoProvider, + String dataSource, + int numThreads, + Duration httpTimeout, + long numRetries + ) + { + super(httpClient, jsonMapper, taskInfoProvider, httpTimeout, dataSource, numThreads, numRetries); + } + + public boolean stop(final String id, final boolean publish) + { + log.debug("Stop task[%s] publish[%s]", id, publish); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent( + id, + HttpMethod.POST, + "stop", + publish ? "publish=true" : null, + true + ); + return isSuccess(response); + } + catch (NoTaskLocationException e) { + return false; + } + catch (TaskNotRunnableException e) { + log.info("Task [%s] couldn't be stopped because it is no longer running", id); + return true; + } + catch (Exception e) { + log.warn(e, "Exception while stopping task [%s]", id); + return false; + } + } + + public boolean resume(final String id) + { + log.debug("Resume task[%s]", id); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent(id, HttpMethod.POST, "resume", null, true); + return isSuccess(response); + } + catch (NoTaskLocationException | IOException e) { + log.warn(e, "Exception while stopping task [%s]", id); + return false; + } + } + + + public Map<PartitionType, SequenceType> pause(final String id) + { + log.debug("Pause task[%s]", id); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent( + id, + HttpMethod.POST, + "pause", + null, + true + ); + + if (response.getStatus().equals(HttpResponseStatus.OK)) { + log.info("Task [%s] paused successfully", id); + return deserialize(response.getContent(), constructPartitionOffsetMapType(Map.class)); + } + + while (true) { + if (getStatus(id) == SeekableStreamIndexTaskRunner.Status.PAUSED) { + return getCurrentOffsets(id, true); + } + + final Duration delay = newRetryPolicy().getAndIncrementRetryDelay(); + if (delay == null) { + log.error("Task [%s] failed to pause, aborting", id); + throw new ISE("Task [%s] failed to pause, aborting", id); + } else { + final long sleepTime = delay.getMillis(); + log.info( + "Still waiting for task [%s] to pause; will try again in [%s]", + id, + new Duration(sleepTime).toString() + ); + Thread.sleep(sleepTime); + } + } + } + catch (NoTaskLocationException e) { + log.error("Exception [%s] while pausing Task [%s]", e.getMessage(), id); + return ImmutableMap.of(); + } + catch (IOException | InterruptedException e) { + throw new RE( + StringUtils.format("Exception [%s] while pausing Task [%s]", e.getMessage(), id), + e + ); + } + } + + public SeekableStreamIndexTaskRunner.Status getStatus(final String id) + { + log.debug("GetStatus task[%s]", id); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent(id, HttpMethod.GET, "status", null, true); + return deserialize(response.getContent(), SeekableStreamIndexTaskRunner.Status.class); + } + catch (NoTaskLocationException e) { + return SeekableStreamIndexTaskRunner.Status.NOT_STARTED; + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + + @Nullable + public DateTime getStartTime(final String id) + { + log.debug("GetStartTime task[%s]", id); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent(id, HttpMethod.GET, "time/start", null, true); + return response.getContent() == null || response.getContent().isEmpty() + ? null + : deserialize(response.getContent(), DateTime.class); + } + catch (NoTaskLocationException e) { + return null; + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public Map<String, Object> getMovingAverages(final String id) + { + log.debug("GetMovingAverages task[%s]", id); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent( + id, + HttpMethod.GET, + "rowStats", + null, + true + ); + return response.getContent() == null || response.getContent().isEmpty() + ? Collections.emptyMap() + : deserialize(response.getContent(), JacksonUtils.TYPE_REFERENCE_MAP_STRING_OBJECT); + } + catch (NoTaskLocationException e) { + return Collections.emptyMap(); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public Map<PartitionType, SequenceType> getCurrentOffsets(final String id, final boolean retry) + { + log.debug("GetCurrentOffsets task[%s] retry[%s]", id, retry); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent( + id, + HttpMethod.GET, + "offsets/current", + null, + retry + ); + return deserialize(response.getContent(), constructPartitionOffsetMapType(Map.class)); + } + catch (NoTaskLocationException e) { + return ImmutableMap.of(); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public TreeMap<Integer, Map<PartitionType, SequenceType>> getCheckpoints(final String id, final boolean retry) + { + log.debug("GetCheckpoints task[%s] retry[%s]", id, retry); + try { + final FullResponseHolder response = submitRequestWithEmptyContent(id, HttpMethod.GET, "checkpoints", null, retry); + return deserialize( + response.getContent(), + constructCheckpointMapType() + ); + } + catch (NoTaskLocationException e) { + return new TreeMap<>(); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public ListenableFuture<TreeMap<Integer, Map<PartitionType, SequenceType>>> getCheckpointsAsync( + final String id, + final boolean retry + ) + { + return doAsync(() -> getCheckpoints(id, retry)); + } + + public Map<PartitionType, SequenceType> getEndOffsets(final String id) + { + log.debug("GetEndOffsets task[%s]", id); + + try { + final FullResponseHolder response = submitRequestWithEmptyContent(id, HttpMethod.GET, "offsets/end", null, true); + return deserialize(response.getContent(), constructPartitionOffsetMapType(Map.class)); + } + catch (NoTaskLocationException e) { + return ImmutableMap.of(); + } + catch (IOException e) { + throw new RuntimeException(e); + } + } + + public boolean setEndOffsets( + final String id, + final Map<PartitionType, SequenceType> endOffsets, + final boolean finalize + ) throws IOException + { + log.debug("SetEndOffsets task[%s] endOffsets[%s] finalize[%s]", id, endOffsets, finalize); + + try { + final FullResponseHolder response = submitJsonRequest( + id, + HttpMethod.POST, + "offsets/end", + StringUtils.format("finish=%s", finalize), + serialize(endOffsets), + true + ); + return isSuccess(response); + } + catch (NoTaskLocationException e) { + return false; + } + } + + public ListenableFuture<Boolean> stopAsync(final String id, final boolean publish) + { + return doAsync(() -> stop(id, publish)); + } + + + public ListenableFuture<Boolean> resumeAsync(final String id) + { + return doAsync(() -> resume(id)); + } + + + public ListenableFuture<DateTime> getStartTimeAsync(final String id) + { + return doAsync(() -> getStartTime(id)); + } + + + public ListenableFuture<Map<PartitionType, SequenceType>> pauseAsync(final String id) + { + return doAsync(() -> pause(id)); + } + + public ListenableFuture<Boolean> setEndOffsetsAsync( + final String id, + final Map<PartitionType, SequenceType> endOffsets, + final boolean finalize + ) + { + return doAsync(() -> setEndOffsets(id, endOffsets, finalize)); + } + + public ListenableFuture<Map<PartitionType, SequenceType>> getCurrentOffsetsAsync(final String id, final boolean retry) + { + return doAsync(() -> getCurrentOffsets(id, retry)); + } + + public ListenableFuture<Map<PartitionType, SequenceType>> getEndOffsetsAsync(final String id) + { + return doAsync(() -> getEndOffsets(id)); + } + + + public ListenableFuture<Map<String, Object>> getMovingAveragesAsync(final String id) + { + return doAsync(() -> getMovingAverages(id)); + } + + + public ListenableFuture<SeekableStreamIndexTaskRunner.Status> getStatusAsync(final String id) + { + return doAsync(() -> getStatus(id)); + } + + private JavaType constructCheckpointMapType() + { + ObjectMapper mapper = new ObjectMapper(); + return mapper.getTypeFactory() + .constructMapType( + TreeMap.class, + mapper.getTypeFactory().constructType(Integer.class), + constructPartitionOffsetMapType(TreeMap.class) + ); + } + + protected abstract JavaType constructPartitionOffsetMapType(Class<? extends Map> mapType); Review comment: Since `objectMapper` is private in `IndexTaskClient` and not accessible from the child classes, you added a new objectMapper [here](https://github.com/apache/incubator-druid/pull/6431/files#diff-edc0abdecae99c05a70ae5436d115827R34). However, I don't think this is a good idea. Instead of this method, I would suggest to add two methods to get the `Class` of `PartitionType` and `SequenceType` and add `deserializeMap` method to `IndexTaskClient` like below. in SeekableStreamIndexTaskClient, ``` protected abstract Class<PartitionType> getPartitionType(); protected abstract Class<SequenceType> getSequenceType(); ``` In IndexTaskClient, ``` protected <T> T deserializeMap(String content, Class<? extends Map> mapClass, Class<?> keyClass, Class<?> valueClass) throws IOException { return deserialize(content, objectMapper.getTypeFactory().constructMapType(mapClass, keyClass, valueClass)); } ``` ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
