Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.io.asyncfs;
019
020import static org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputSaslHelper.createEncryptor;
021import static org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutputSaslHelper.trySaslNegotiate;
022import static org.apache.hadoop.hbase.util.LocatedBlockHelper.getLocatedBlockLocations;
023import static org.apache.hadoop.hbase.util.NettyFutureUtils.addListener;
024import static org.apache.hadoop.hbase.util.NettyFutureUtils.safeClose;
025import static org.apache.hadoop.hbase.util.NettyFutureUtils.safeWriteAndFlush;
026import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_CLIENT_SOCKET_TIMEOUT_KEY;
027import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_CLIENT_USE_DN_HOSTNAME;
028import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_CLIENT_USE_DN_HOSTNAME_DEFAULT;
029import static org.apache.hadoop.hdfs.protocol.datatransfer.BlockConstructionStage.PIPELINE_SETUP_CREATE;
030import static org.apache.hbase.thirdparty.io.netty.channel.ChannelOption.CONNECT_TIMEOUT_MILLIS;
031import static org.apache.hbase.thirdparty.io.netty.handler.timeout.IdleState.READER_IDLE;
032
033import java.io.IOException;
034import java.io.InterruptedIOException;
035import java.lang.reflect.InvocationTargetException;
036import java.lang.reflect.Method;
037import java.util.ArrayList;
038import java.util.Collection;
039import java.util.EnumSet;
040import java.util.HashSet;
041import java.util.IdentityHashMap;
042import java.util.List;
043import java.util.Map;
044import java.util.Set;
045import java.util.concurrent.TimeUnit;
046import java.util.stream.Collectors;
047import org.apache.hadoop.conf.Configuration;
048import org.apache.hadoop.crypto.CryptoProtocolVersion;
049import org.apache.hadoop.crypto.Encryptor;
050import org.apache.hadoop.fs.CreateFlag;
051import org.apache.hadoop.fs.FileSystem;
052import org.apache.hadoop.fs.FileSystemLinkResolver;
053import org.apache.hadoop.fs.Path;
054import org.apache.hadoop.fs.StorageType;
055import org.apache.hadoop.fs.UnresolvedLinkException;
056import org.apache.hadoop.fs.permission.FsPermission;
057import org.apache.hadoop.hbase.client.ConnectionUtils;
058import org.apache.hadoop.hbase.io.asyncfs.monitor.ExcludeDatanodeManager;
059import org.apache.hadoop.hbase.io.asyncfs.monitor.StreamSlowMonitor;
060import org.apache.hadoop.hbase.util.CancelableProgressable;
061import org.apache.hadoop.hdfs.DFSClient;
062import org.apache.hadoop.hdfs.DFSOutputStream;
063import org.apache.hadoop.hdfs.DistributedFileSystem;
064import org.apache.hadoop.hdfs.protocol.ClientProtocol;
065import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
066import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
067import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
068import org.apache.hadoop.hdfs.protocol.LocatedBlock;
069import org.apache.hadoop.hdfs.protocol.datatransfer.BlockConstructionStage;
070import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtoUtil;
071import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtocol;
072import org.apache.hadoop.hdfs.protocol.datatransfer.Op;
073import org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck;
074import org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck.ECN;
075import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BaseHeaderProto;
076import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
077import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.CachingStrategyProto;
078import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ChecksumProto;
079import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ClientOperationHeaderProto;
080import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpWriteBlockProto;
081import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.PipelineAckProto;
082import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
083import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
084import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
085import org.apache.hadoop.hdfs.security.token.block.InvalidBlockTokenException;
086import org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException;
087import org.apache.hadoop.io.EnumSetWritable;
088import org.apache.hadoop.ipc.RemoteException;
089import org.apache.hadoop.net.NetUtils;
090import org.apache.hadoop.security.token.Token;
091import org.apache.hadoop.util.DataChecksum;
092import org.apache.yetus.audience.InterfaceAudience;
093import org.slf4j.Logger;
094import org.slf4j.LoggerFactory;
095
096import org.apache.hbase.thirdparty.com.google.protobuf.CodedOutputStream;
097import org.apache.hbase.thirdparty.io.netty.bootstrap.Bootstrap;
098import org.apache.hbase.thirdparty.io.netty.buffer.ByteBuf;
099import org.apache.hbase.thirdparty.io.netty.buffer.ByteBufAllocator;
100import org.apache.hbase.thirdparty.io.netty.buffer.ByteBufOutputStream;
101import org.apache.hbase.thirdparty.io.netty.buffer.PooledByteBufAllocator;
102import org.apache.hbase.thirdparty.io.netty.channel.Channel;
103import org.apache.hbase.thirdparty.io.netty.channel.ChannelFuture;
104import org.apache.hbase.thirdparty.io.netty.channel.ChannelFutureListener;
105import org.apache.hbase.thirdparty.io.netty.channel.ChannelHandler;
106import org.apache.hbase.thirdparty.io.netty.channel.ChannelHandlerContext;
107import org.apache.hbase.thirdparty.io.netty.channel.ChannelInitializer;
108import org.apache.hbase.thirdparty.io.netty.channel.ChannelPipeline;
109import org.apache.hbase.thirdparty.io.netty.channel.EventLoop;
110import org.apache.hbase.thirdparty.io.netty.channel.EventLoopGroup;
111import org.apache.hbase.thirdparty.io.netty.channel.SimpleChannelInboundHandler;
112import org.apache.hbase.thirdparty.io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
113import org.apache.hbase.thirdparty.io.netty.handler.timeout.IdleStateEvent;
114import org.apache.hbase.thirdparty.io.netty.handler.timeout.IdleStateHandler;
115import org.apache.hbase.thirdparty.io.netty.util.concurrent.Future;
116import org.apache.hbase.thirdparty.io.netty.util.concurrent.FutureListener;
117import org.apache.hbase.thirdparty.io.netty.util.concurrent.Promise;
118
119/**
120 * Helper class for implementing {@link FanOutOneBlockAsyncDFSOutput}.
121 */
122@InterfaceAudience.Private
123public final class FanOutOneBlockAsyncDFSOutputHelper {
124  private static final Logger LOG =
125    LoggerFactory.getLogger(FanOutOneBlockAsyncDFSOutputHelper.class);
126
127  private FanOutOneBlockAsyncDFSOutputHelper() {
128  }
129
130  public static final String ASYNC_DFS_OUTPUT_CREATE_MAX_RETRIES = "hbase.fs.async.create.retries";
131
132  public static final int DEFAULT_ASYNC_DFS_OUTPUT_CREATE_MAX_RETRIES = 10;
133  // use pooled allocator for performance.
134  private static final ByteBufAllocator ALLOC = PooledByteBufAllocator.DEFAULT;
135
136  // copied from DFSPacket since it is package private.
137  public static final long HEART_BEAT_SEQNO = -1L;
138
139  // Timeouts for communicating with DataNode for streaming writes/reads
140  public static final int READ_TIMEOUT = 60 * 1000;
141
142  private interface LeaseManager {
143
144    void begin(FanOutOneBlockAsyncDFSOutput output);
145
146    void end(FanOutOneBlockAsyncDFSOutput output);
147  }
148
149  private static final LeaseManager LEASE_MANAGER;
150
151  // helper class for creating files.
152  private interface FileCreator {
153    default HdfsFileStatus create(ClientProtocol instance, String src, FsPermission masked,
154      String clientName, EnumSetWritable<CreateFlag> flag, boolean createParent, short replication,
155      long blockSize, CryptoProtocolVersion[] supportedVersions) throws Exception {
156      try {
157        return (HdfsFileStatus) createObject(instance, src, masked, clientName, flag, createParent,
158          replication, blockSize, supportedVersions);
159      } catch (InvocationTargetException e) {
160        if (e.getCause() instanceof Exception) {
161          throw (Exception) e.getCause();
162        } else {
163          throw new RuntimeException(e.getCause());
164        }
165      }
166    }
167
168    Object createObject(ClientProtocol instance, String src, FsPermission masked, String clientName,
169      EnumSetWritable<CreateFlag> flag, boolean createParent, short replication, long blockSize,
170      CryptoProtocolVersion[] supportedVersions) throws Exception;
171  }
172
173  private static final FileCreator FILE_CREATOR;
174
175  private static LeaseManager createLeaseManager3_4() throws NoSuchMethodException {
176    Method beginFileLeaseMethod =
177      DFSClient.class.getDeclaredMethod("beginFileLease", String.class, DFSOutputStream.class);
178    beginFileLeaseMethod.setAccessible(true);
179    Method endFileLeaseMethod = DFSClient.class.getDeclaredMethod("endFileLease", String.class);
180    endFileLeaseMethod.setAccessible(true);
181    Method getUniqKeyMethod = DFSOutputStream.class.getMethod("getUniqKey");
182    return new LeaseManager() {
183
184      private String getUniqKey(FanOutOneBlockAsyncDFSOutput output)
185        throws IllegalAccessException, InvocationTargetException {
186        return (String) getUniqKeyMethod.invoke(output.getDummyStream());
187      }
188
189      @Override
190      public void begin(FanOutOneBlockAsyncDFSOutput output) {
191        try {
192          beginFileLeaseMethod.invoke(output.getClient(), getUniqKey(output),
193            output.getDummyStream());
194        } catch (IllegalAccessException | InvocationTargetException e) {
195          throw new RuntimeException(e);
196        }
197      }
198
199      @Override
200      public void end(FanOutOneBlockAsyncDFSOutput output) {
201        try {
202          endFileLeaseMethod.invoke(output.getClient(), getUniqKey(output));
203        } catch (IllegalAccessException | InvocationTargetException e) {
204          throw new RuntimeException(e);
205        }
206      }
207    };
208  }
209
210  private static LeaseManager createLeaseManager3() throws NoSuchMethodException {
211    Method beginFileLeaseMethod =
212      DFSClient.class.getDeclaredMethod("beginFileLease", long.class, DFSOutputStream.class);
213    beginFileLeaseMethod.setAccessible(true);
214    Method endFileLeaseMethod = DFSClient.class.getDeclaredMethod("endFileLease", long.class);
215    endFileLeaseMethod.setAccessible(true);
216    return new LeaseManager() {
217
218      @Override
219      public void begin(FanOutOneBlockAsyncDFSOutput output) {
220        try {
221          beginFileLeaseMethod.invoke(output.getClient(), output.getDummyStream().getFileId(),
222            output.getDummyStream());
223        } catch (IllegalAccessException | InvocationTargetException e) {
224          throw new RuntimeException(e);
225        }
226      }
227
228      @Override
229      public void end(FanOutOneBlockAsyncDFSOutput output) {
230        try {
231          endFileLeaseMethod.invoke(output.getClient(), output.getDummyStream().getFileId());
232        } catch (IllegalAccessException | InvocationTargetException e) {
233          throw new RuntimeException(e);
234        }
235      }
236    };
237  }
238
239  private static LeaseManager createLeaseManager() throws NoSuchMethodException {
240    try {
241      return createLeaseManager3_4();
242    } catch (NoSuchMethodException e) {
243      LOG.debug("DFSClient::beginFileLease wrong arguments, should be hadoop 3.3 or below");
244    }
245
246    return createLeaseManager3();
247  }
248
249  private static FileCreator createFileCreator3_3() throws NoSuchMethodException {
250    Method createMethod = ClientProtocol.class.getMethod("create", String.class, FsPermission.class,
251      String.class, EnumSetWritable.class, boolean.class, short.class, long.class,
252      CryptoProtocolVersion[].class, String.class, String.class);
253
254    return (instance, src, masked, clientName, flag, createParent, replication, blockSize,
255      supportedVersions) -> {
256      return (HdfsFileStatus) createMethod.invoke(instance, src, masked, clientName, flag,
257        createParent, replication, blockSize, supportedVersions, null, null);
258    };
259  }
260
261  private static FileCreator createFileCreator3() throws NoSuchMethodException {
262    Method createMethod = ClientProtocol.class.getMethod("create", String.class, FsPermission.class,
263      String.class, EnumSetWritable.class, boolean.class, short.class, long.class,
264      CryptoProtocolVersion[].class, String.class);
265
266    return (instance, src, masked, clientName, flag, createParent, replication, blockSize,
267      supportedVersions) -> {
268      return (HdfsFileStatus) createMethod.invoke(instance, src, masked, clientName, flag,
269        createParent, replication, blockSize, supportedVersions, null);
270    };
271  }
272
273  private static FileCreator createFileCreator() throws NoSuchMethodException {
274    try {
275      return createFileCreator3_3();
276    } catch (NoSuchMethodException e) {
277      LOG.debug("ClientProtocol::create wrong number of arguments, should be hadoop 3.2 or below");
278    }
279
280    return createFileCreator3();
281  }
282
283  // cancel the processing if DFSClient is already closed.
284  static final class CancelOnClose implements CancelableProgressable {
285
286    private final DFSClient client;
287
288    public CancelOnClose(DFSClient client) {
289      this.client = client;
290    }
291
292    @Override
293    public boolean progress() {
294      return client.isClientRunning();
295    }
296  }
297
298  static {
299    try {
300      LEASE_MANAGER = createLeaseManager();
301      FILE_CREATOR = createFileCreator();
302    } catch (Exception e) {
303      String msg = "Couldn't properly initialize access to HDFS internals. Please "
304        + "update your WAL Provider to not make use of the 'asyncfs' provider. See "
305        + "HBASE-16110 for more information.";
306      LOG.error(msg, e);
307      throw new Error(msg, e);
308    }
309  }
310
311  private static void beginFileLease(FanOutOneBlockAsyncDFSOutput output) {
312    LEASE_MANAGER.begin(output);
313  }
314
315  static void endFileLease(FanOutOneBlockAsyncDFSOutput output) {
316    LEASE_MANAGER.end(output);
317  }
318
319  static DataChecksum createChecksum(DFSClient client) {
320    return client.getConf().createChecksum(null);
321  }
322
323  static Status getStatus(PipelineAckProto ack) {
324    List<Integer> flagList = ack.getFlagList();
325    Integer headerFlag;
326    if (flagList.isEmpty()) {
327      Status reply = ack.getReply(0);
328      headerFlag = PipelineAck.combineHeader(ECN.DISABLED, reply);
329    } else {
330      headerFlag = flagList.get(0);
331    }
332    return PipelineAck.getStatusFromHeader(headerFlag);
333  }
334
335  private static void processWriteBlockResponse(Channel channel, DatanodeInfo dnInfo,
336    Promise<Channel> promise, int timeoutMs) {
337    channel.pipeline().addLast(new IdleStateHandler(timeoutMs, 0, 0, TimeUnit.MILLISECONDS),
338      new ProtobufVarint32FrameDecoder(),
339      new ProtobufDecoder(BlockOpResponseProto.getDefaultInstance()),
340      new SimpleChannelInboundHandler<BlockOpResponseProto>() {
341
342        @Override
343        protected void channelRead0(ChannelHandlerContext ctx, BlockOpResponseProto resp)
344          throws Exception {
345          Status pipelineStatus = resp.getStatus();
346          if (PipelineAck.isRestartOOBStatus(pipelineStatus)) {
347            throw new IOException("datanode " + dnInfo + " is restarting");
348          }
349          String logInfo = "ack with firstBadLink as " + resp.getFirstBadLink();
350          if (resp.getStatus() != Status.SUCCESS) {
351            if (resp.getStatus() == Status.ERROR_ACCESS_TOKEN) {
352              throw new InvalidBlockTokenException("Got access token error" + ", status message "
353                + resp.getMessage() + ", " + logInfo);
354            } else {
355              throw new IOException("Got error" + ", status=" + resp.getStatus().name()
356                + ", status message " + resp.getMessage() + ", " + logInfo);
357            }
358          }
359          // success
360          ChannelPipeline p = ctx.pipeline();
361          for (ChannelHandler handler; (handler = p.removeLast()) != null;) {
362            // do not remove all handlers because we may have wrap or unwrap handlers at the header
363            // of pipeline.
364            if (handler instanceof IdleStateHandler) {
365              break;
366            }
367          }
368          // Disable auto read here. Enable it after we setup the streaming pipeline in
369          // FanOutOneBLockAsyncDFSOutput.
370          ctx.channel().config().setAutoRead(false);
371          promise.trySuccess(ctx.channel());
372        }
373
374        @Override
375        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
376          promise.tryFailure(new IOException("connection to " + dnInfo + " is closed"));
377        }
378
379        @Override
380        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
381          if (evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state() == READER_IDLE) {
382            promise
383              .tryFailure(new IOException("Timeout(" + timeoutMs + "ms) waiting for response"));
384          } else {
385            super.userEventTriggered(ctx, evt);
386          }
387        }
388
389        @Override
390        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
391          promise.tryFailure(cause);
392        }
393      });
394  }
395
396  private static void requestWriteBlock(Channel channel, StorageType storageType,
397    OpWriteBlockProto.Builder writeBlockProtoBuilder) throws IOException {
398    OpWriteBlockProto proto =
399      writeBlockProtoBuilder.setStorageType(PBHelperClient.convertStorageType(storageType)).build();
400    int protoLen = proto.getSerializedSize();
401    ByteBuf buffer =
402      channel.alloc().buffer(3 + CodedOutputStream.computeUInt32SizeNoTag(protoLen) + protoLen);
403    buffer.writeShort(DataTransferProtocol.DATA_TRANSFER_VERSION);
404    buffer.writeByte(Op.WRITE_BLOCK.code);
405    proto.writeDelimitedTo(new ByteBufOutputStream(buffer));
406    safeWriteAndFlush(channel, buffer);
407  }
408
409  private static void initialize(Configuration conf, Channel channel, DatanodeInfo dnInfo,
410    StorageType storageType, OpWriteBlockProto.Builder writeBlockProtoBuilder, int timeoutMs,
411    DFSClient client, Token<BlockTokenIdentifier> accessToken, Promise<Channel> promise)
412    throws IOException {
413    Promise<Void> saslPromise = channel.eventLoop().newPromise();
414    trySaslNegotiate(conf, channel, dnInfo, timeoutMs, client, accessToken, saslPromise);
415    addListener(saslPromise, new FutureListener<Void>() {
416
417      @Override
418      public void operationComplete(Future<Void> future) throws Exception {
419        if (future.isSuccess()) {
420          // setup response processing pipeline first, then send request.
421          processWriteBlockResponse(channel, dnInfo, promise, timeoutMs);
422          requestWriteBlock(channel, storageType, writeBlockProtoBuilder);
423        } else {
424          promise.tryFailure(future.cause());
425        }
426      }
427    });
428  }
429
430  private static List<Future<Channel>> connectToDataNodes(Configuration conf, DFSClient client,
431    String clientName, LocatedBlock locatedBlock, long maxBytesRcvd, long latestGS,
432    BlockConstructionStage stage, DataChecksum summer, EventLoopGroup eventLoopGroup,
433    Class<? extends Channel> channelClass) {
434    StorageType[] storageTypes = locatedBlock.getStorageTypes();
435    DatanodeInfo[] datanodeInfos = getLocatedBlockLocations(locatedBlock);
436    boolean connectToDnViaHostname =
437      conf.getBoolean(DFS_CLIENT_USE_DN_HOSTNAME, DFS_CLIENT_USE_DN_HOSTNAME_DEFAULT);
438    int timeoutMs = conf.getInt(DFS_CLIENT_SOCKET_TIMEOUT_KEY, READ_TIMEOUT);
439    ExtendedBlock blockCopy = new ExtendedBlock(locatedBlock.getBlock());
440    blockCopy.setNumBytes(locatedBlock.getBlockSize());
441    ClientOperationHeaderProto header = ClientOperationHeaderProto.newBuilder()
442      .setBaseHeader(BaseHeaderProto.newBuilder().setBlock(PBHelperClient.convert(blockCopy))
443        .setToken(PBHelperClient.convert(locatedBlock.getBlockToken())))
444      .setClientName(clientName).build();
445    ChecksumProto checksumProto = DataTransferProtoUtil.toProto(summer);
446    OpWriteBlockProto.Builder writeBlockProtoBuilder =
447      OpWriteBlockProto.newBuilder().setHeader(header)
448        .setStage(OpWriteBlockProto.BlockConstructionStage.valueOf(stage.name())).setPipelineSize(1)
449        .setMinBytesRcvd(locatedBlock.getBlock().getNumBytes()).setMaxBytesRcvd(maxBytesRcvd)
450        .setLatestGenerationStamp(latestGS).setRequestedChecksum(checksumProto)
451        .setCachingStrategy(CachingStrategyProto.newBuilder().setDropBehind(true).build());
452    List<Future<Channel>> futureList = new ArrayList<>(datanodeInfos.length);
453    for (int i = 0; i < datanodeInfos.length; i++) {
454      DatanodeInfo dnInfo = datanodeInfos[i];
455      StorageType storageType = storageTypes[i];
456      Promise<Channel> promise = eventLoopGroup.next().newPromise();
457      futureList.add(promise);
458      String dnAddr = dnInfo.getXferAddr(connectToDnViaHostname);
459      addListener(new Bootstrap().group(eventLoopGroup).channel(channelClass)
460        .option(CONNECT_TIMEOUT_MILLIS, timeoutMs).handler(new ChannelInitializer<Channel>() {
461
462          @Override
463          protected void initChannel(Channel ch) throws Exception {
464            // we need to get the remote address of the channel so we can only move on after
465            // channel connected. Leave an empty implementation here because netty does not allow
466            // a null handler.
467          }
468        }).connect(NetUtils.createSocketAddr(dnAddr)), new ChannelFutureListener() {
469
470          @Override
471          public void operationComplete(ChannelFuture future) throws Exception {
472            if (future.isSuccess()) {
473              initialize(conf, future.channel(), dnInfo, storageType, writeBlockProtoBuilder,
474                timeoutMs, client, locatedBlock.getBlockToken(), promise);
475            } else {
476              promise.tryFailure(future.cause());
477            }
478          }
479        });
480    }
481    return futureList;
482  }
483
484  /**
485   * Exception other than RemoteException thrown when calling create on namenode
486   */
487  public static class NameNodeException extends IOException {
488
489    private static final long serialVersionUID = 3143237406477095390L;
490
491    public NameNodeException(Throwable cause) {
492      super(cause);
493    }
494  }
495
496  private static EnumSetWritable<CreateFlag> getCreateFlags(boolean overwrite,
497    boolean noLocalWrite) {
498    List<CreateFlag> flags = new ArrayList<>();
499    flags.add(CreateFlag.CREATE);
500    if (overwrite) {
501      flags.add(CreateFlag.OVERWRITE);
502    }
503    if (noLocalWrite) {
504      flags.add(CreateFlag.NO_LOCAL_WRITE);
505    }
506    flags.add(CreateFlag.SHOULD_REPLICATE);
507    return new EnumSetWritable<>(EnumSet.copyOf(flags));
508  }
509
510  private static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem dfs, String src,
511    boolean overwrite, boolean createParent, short replication, long blockSize,
512    EventLoopGroup eventLoopGroup, Class<? extends Channel> channelClass, StreamSlowMonitor monitor,
513    boolean noLocalWrite) throws IOException {
514    Configuration conf = dfs.getConf();
515    DFSClient client = dfs.getClient();
516    String clientName = client.getClientName();
517    ClientProtocol namenode = client.getNamenode();
518    int createMaxRetries =
519      conf.getInt(ASYNC_DFS_OUTPUT_CREATE_MAX_RETRIES, DEFAULT_ASYNC_DFS_OUTPUT_CREATE_MAX_RETRIES);
520    ExcludeDatanodeManager excludeDatanodeManager = monitor.getExcludeDatanodeManager();
521    Set<DatanodeInfo> toExcludeNodes =
522      new HashSet<>(excludeDatanodeManager.getExcludeDNs().keySet());
523    for (int retry = 0;; retry++) {
524      if (LOG.isDebugEnabled()) {
525        LOG.debug("When create output stream for {}, exclude list is {}, retry={}", src,
526          getDataNodeInfo(toExcludeNodes), retry);
527      }
528      EnumSetWritable<CreateFlag> createFlags = getCreateFlags(overwrite, noLocalWrite);
529      HdfsFileStatus stat;
530      try {
531        stat = FILE_CREATOR.create(namenode, src,
532          FsPermission.getFileDefault().applyUMask(FsPermission.getUMask(conf)), clientName,
533          createFlags, createParent, replication, blockSize, CryptoProtocolVersion.supported());
534      } catch (Exception e) {
535        if (e instanceof RemoteException) {
536          throw (RemoteException) e;
537        } else {
538          throw new NameNodeException(e);
539        }
540      }
541      boolean succ = false;
542      LocatedBlock locatedBlock = null;
543      List<Future<Channel>> futureList = null;
544      try {
545        DataChecksum summer = createChecksum(client);
546        locatedBlock = namenode.addBlock(src, client.getClientName(), null,
547          toExcludeNodes.toArray(new DatanodeInfo[0]), stat.getFileId(), null, null);
548        Map<Channel, DatanodeInfo> datanodes = new IdentityHashMap<>();
549        futureList = connectToDataNodes(conf, client, clientName, locatedBlock, 0L, 0L,
550          PIPELINE_SETUP_CREATE, summer, eventLoopGroup, channelClass);
551        for (int i = 0, n = futureList.size(); i < n; i++) {
552          DatanodeInfo datanodeInfo = getLocatedBlockLocations(locatedBlock)[i];
553          try {
554            datanodes.put(futureList.get(i).syncUninterruptibly().getNow(), datanodeInfo);
555          } catch (Exception e) {
556            // exclude the broken DN next time
557            toExcludeNodes.add(datanodeInfo);
558            excludeDatanodeManager.tryAddExcludeDN(datanodeInfo, "connect error");
559            throw e;
560          }
561        }
562        Encryptor encryptor = createEncryptor(conf, stat, client);
563        FanOutOneBlockAsyncDFSOutput output =
564          new FanOutOneBlockAsyncDFSOutput(conf, dfs, client, namenode, clientName, src, stat,
565            createFlags.get(), locatedBlock, encryptor, datanodes, summer, ALLOC, monitor);
566        beginFileLease(output);
567        succ = true;
568        return output;
569      } catch (RemoteException e) {
570        LOG.warn("create fan-out dfs output {} failed, retry = {}", src, retry, e);
571        if (shouldRetryCreate(e)) {
572          if (retry >= createMaxRetries) {
573            throw e.unwrapRemoteException();
574          }
575        } else {
576          throw e.unwrapRemoteException();
577        }
578      } catch (IOException e) {
579        LOG.warn("create fan-out dfs output {} failed, retry = {}", src, retry, e);
580        if (retry >= createMaxRetries) {
581          throw e;
582        }
583        // overwrite the old broken file.
584        overwrite = true;
585        try {
586          Thread.sleep(ConnectionUtils.getPauseTime(100, retry));
587        } catch (InterruptedException ie) {
588          throw new InterruptedIOException();
589        }
590      } finally {
591        if (!succ) {
592          if (futureList != null) {
593            for (Future<Channel> f : futureList) {
594              addListener(f, new FutureListener<Channel>() {
595
596                @Override
597                public void operationComplete(Future<Channel> future) throws Exception {
598                  if (future.isSuccess()) {
599                    safeClose(future.getNow());
600                  }
601                }
602              });
603            }
604          }
605        }
606      }
607    }
608  }
609
610  /**
611   * Create a {@link FanOutOneBlockAsyncDFSOutput}. The method maybe blocked so do not call it
612   * inside an {@link EventLoop}.
613   */
614  public static FanOutOneBlockAsyncDFSOutput createOutput(DistributedFileSystem dfs, Path f,
615    boolean overwrite, boolean createParent, short replication, long blockSize,
616    EventLoopGroup eventLoopGroup, Class<? extends Channel> channelClass,
617    final StreamSlowMonitor monitor, boolean noLocalWrite) throws IOException {
618    return new FileSystemLinkResolver<FanOutOneBlockAsyncDFSOutput>() {
619
620      @Override
621      public FanOutOneBlockAsyncDFSOutput doCall(Path p)
622        throws IOException, UnresolvedLinkException {
623        return createOutput(dfs, p.toUri().getPath(), overwrite, createParent, replication,
624          blockSize, eventLoopGroup, channelClass, monitor, noLocalWrite);
625      }
626
627      @Override
628      public FanOutOneBlockAsyncDFSOutput next(FileSystem fs, Path p) throws IOException {
629        throw new UnsupportedOperationException();
630      }
631    }.resolve(dfs, f);
632  }
633
634  public static boolean shouldRetryCreate(RemoteException e) {
635    // RetryStartFileException is introduced in HDFS 2.6+, so here we can only use the class name.
636    // For exceptions other than this, we just throw it out. This is same with
637    // DFSOutputStream.newStreamForCreate.
638    return e.getClassName().endsWith("RetryStartFileException");
639  }
640
641  static void completeFile(FanOutOneBlockAsyncDFSOutput output, DFSClient client,
642    ClientProtocol namenode, String src, String clientName, ExtendedBlock block,
643    HdfsFileStatus stat) {
644    for (int retry = 0;; retry++) {
645      try {
646        if (namenode.complete(src, clientName, block, stat.getFileId())) {
647          endFileLease(output);
648          return;
649        } else {
650          LOG.warn("complete file " + src + " not finished, retry = " + retry);
651        }
652      } catch (RemoteException e) {
653        IOException ioe = e.unwrapRemoteException();
654        if (ioe instanceof LeaseExpiredException) {
655          LOG.warn("lease for file " + src + " is expired, give up", e);
656          return;
657        } else {
658          LOG.warn("complete file " + src + " failed, retry = " + retry, e);
659        }
660      } catch (Exception e) {
661        LOG.warn("complete file " + src + " failed, retry = " + retry, e);
662      }
663      sleepIgnoreInterrupt(retry);
664    }
665  }
666
667  static void sleepIgnoreInterrupt(int retry) {
668    try {
669      Thread.sleep(ConnectionUtils.getPauseTime(100, retry));
670    } catch (InterruptedException e) {
671    }
672  }
673
674  public static String getDataNodeInfo(Collection<DatanodeInfo> datanodeInfos) {
675    if (datanodeInfos.isEmpty()) {
676      return "[]";
677    }
678    return datanodeInfos.stream()
679      .map(datanodeInfo -> new StringBuilder().append("(").append(datanodeInfo.getHostName())
680        .append("/").append(datanodeInfo.getInfoAddr()).append(":")
681        .append(datanodeInfo.getInfoPort()).append(")").toString())
682      .collect(Collectors.joining(",", "[", "]"));
683  }
684}