[GitHub] carbondata pull request #1654: [CARBONDATA-1856][PARTITION] Support insert/l...

[xubo245 <gi...@git.apache.org>]

Github user xubo245 commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/1654#discussion_r170155099
  
    --- Diff: hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableOutputFormat.java ---
    @@ -339,28 +345,47 @@ private static void setFileHeader(Configuration configuration, CarbonLoadModel m
         model.setCsvHeaderColumns(columns);
       }
     
    -  private static class CarbonRecordWriter extends RecordWriter<NullWritable, StringArrayWritable> {
    +  public static class CarbonRecordWriter extends RecordWriter<NullWritable, StringArrayWritable> {
     
         private CarbonOutputIteratorWrapper iteratorWrapper;
     
         private DataLoadExecutor dataLoadExecutor;
     
    +    private CarbonLoadModel loadModel;
    +
    +    private ExecutorService executorService;
    +
    +    private Future future;
    +
         public CarbonRecordWriter(CarbonOutputIteratorWrapper iteratorWrapper,
    -        DataLoadExecutor dataLoadExecutor) {
    +        DataLoadExecutor dataLoadExecutor, CarbonLoadModel loadModel, Future future,
    +        ExecutorService executorService) {
           this.iteratorWrapper = iteratorWrapper;
           this.dataLoadExecutor = dataLoadExecutor;
    +      this.loadModel = loadModel;
    +      this.executorService = executorService;
    +      this.future = future;
         }
     
    -    @Override
    -    public void write(NullWritable aVoid, StringArrayWritable strings)
    +    @Override public void write(NullWritable aVoid, StringArrayWritable strings)
             throws InterruptedException {
           iteratorWrapper.write(strings.get());
         }
     
    -    @Override
    -    public void close(TaskAttemptContext taskAttemptContext) {
    +    @Override public void close(TaskAttemptContext taskAttemptContext) throws InterruptedException {
           iteratorWrapper.close();
    -      dataLoadExecutor.close();
    +      try {
    +        future.get();
    +      } catch (ExecutionException e) {
    +        throw new InterruptedException(e.getMessage());
    --- End diff --
    
    Why did you throw InterruptedException?


---