not trying to hold leases on WAL files if we are holding them already. #142
+122
−66
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| /** | ||
| * Copyright 2015 Confluent Inc. | ||
|
There was a problem hiding this comment. nit: If you apply any changes based on the comment above, you may change this one too to be |
||
| * | ||
| * Licensed 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 io.confluent.connect.hdfs.wal; | ||
|
|
||
| public class WALConstants { | ||
| protected static final String LEASE_EXCEPTION_CLASS_NAME | ||
| = "org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException"; | ||
| protected static final long MAX_SLEEP_INTERVAL_MS = 16000L; | ||
| protected static final long INITIAL_SLEEP_INTERVAL_MS = 1000L; | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -34,6 +34,7 @@ | |
| import org.apache.hadoop.io.serializer.Deserializer; | ||
| import org.apache.hadoop.io.serializer.SerializationFactory; | ||
| import org.apache.hadoop.io.serializer.Serializer; | ||
| import org.apache.hadoop.ipc.RemoteException; | ||
| import org.apache.hadoop.util.Options; | ||
| import org.apache.hadoop.util.Time; | ||
|
|
||
|
|
@@ -132,42 +133,56 @@ public static class Writer implements Closeable, Syncable { | |
| throw new IllegalArgumentException("file modifier options not compatible with stream"); | ||
| } | ||
|
|
||
| FileSystem fs = null; | ||
| FSDataOutputStream out; | ||
| boolean ownStream = fileOption != null; | ||
|
|
||
| try { | ||
| if (ownStream) { | ||
| Path p = fileOption.getValue(); | ||
| fs = p.getFileSystem(conf); | ||
| int bufferSize = bufferSizeOption == null | ||
| ? getBufferSize(conf) | ||
| : bufferSizeOption.getValue(); | ||
| short replication = replicationOption == null | ||
| ? fs.getDefaultReplication(p) | ||
| : (short) replicationOption.getValue(); | ||
| long blockSize = blockSizeOption == null | ||
| ? fs.getDefaultBlockSize(p) | ||
| : blockSizeOption.getValue(); | ||
|
|
||
| if (appendIfExistsOption != null && appendIfExistsOption.getValue() && fs.exists(p)) { | ||
| // Read the file and verify header details | ||
| try (WALFile.Reader reader = new WALFile.Reader( | ||
| connectorConfig.getHadoopConfiguration(), | ||
| WALFile.Reader.file(p), | ||
| new Reader.OnlyHeaderOption() | ||
| )) { | ||
| if (reader.getVersion() != VERSION[3]) { | ||
| throw new VersionMismatchException(VERSION[3], reader.getVersion()); | ||
| } | ||
| sync = reader.getSync(); | ||
| } | ||
| out = fs.append(p, bufferSize); | ||
| this.appendMode = true; | ||
| } else { | ||
| out = fs.create(p, true, bufferSize, replication, blockSize); | ||
| } | ||
| } else { | ||
| out = streamOption.getValue(); | ||
| } | ||
|
|
||
| init(connectorConfig, out, ownStream); | ||
| } catch (RemoteException re) { | ||
| log.error("Failed creating a WAL Writer: " + re.getMessage()); | ||
|
There was a problem hiding this comment. Can we do instead so the log will include stack trace info? There was a problem hiding this comment. In fact, is the log even going to be useful since it gets logged again by There was a problem hiding this comment. Yes, I was hoping to log a bit more information so that it is easier to read the debug log which is VERY long. So re.getMessage() is the short version that I like :-) If we log the stack trace, the same info will be logged again all the way up at https://github.com/confluentinc/kafka-connect-hdfs/blob/master/src/main/java/io/confluent/connect/hdfs/TopicPartitionWriter.java#L324. |
||
| if (re.getClassName().equals(WALConstants.LEASE_EXCEPTION_CLASS_NAME)) { | ||
| if (fs != null) { | ||
| fs.close(); | ||
|
There was a problem hiding this comment. Does this need protection from a possible There was a problem hiding this comment. Nope. I think IOException will be caught here: https://github.com/confluentinc/kafka-connect-hdfs/blob/master/src/main/java/io/confluent/connect/hdfs/TopicPartitionWriter.java#L324. |
||
| } | ||
| } | ||
| throw re; | ||
| } | ||
|
|
||
| } | ||
|
|
||
| public static Option file(Path value) { | ||
|
|
@@ -394,6 +409,7 @@ public Reader(Configuration conf, Option... opts) throws IOException { | |
| InputStreamOption streamOpt = Options.getOption(InputStreamOption.class, opts); | ||
| LengthOption lenOpt = Options.getOption(LengthOption.class, opts); | ||
| BufferSizeOption bufOpt = Options.getOption(BufferSizeOption.class, opts); | ||
|
|
||
| // check for consistency | ||
| if ((fileOpt == null) == (streamOpt == null)) { | ||
| throw new | ||
|
|
@@ -402,27 +418,40 @@ public Reader(Configuration conf, Option... opts) throws IOException { | |
| if (fileOpt == null && bufOpt != null) { | ||
| throw new IllegalArgumentException("buffer size can only be set when a file is specified."); | ||
| } | ||
|
|
||
| // figure out the real values | ||
| Path filename = null; | ||
| FSDataInputStream file; | ||
| final long len; | ||
| FileSystem fs = null; | ||
|
|
||
| try { | ||
| if (fileOpt != null) { | ||
| filename = fileOpt.getValue(); | ||
| fs = filename.getFileSystem(conf); | ||
| int bufSize = bufOpt == null ? getBufferSize(conf) : bufOpt.getValue(); | ||
| len = null == lenOpt | ||
| ? fs.getFileStatus(filename).getLen() | ||
| : lenOpt.getValue(); | ||
| file = openFile(fs, filename, bufSize, len); | ||
| } else { | ||
| len = null == lenOpt ? Long.MAX_VALUE : lenOpt.getValue(); | ||
| file = streamOpt.getValue(); | ||
| } | ||
| StartOption startOpt = Options.getOption(StartOption.class, opts); | ||
| long start = startOpt == null ? 0 : startOpt.getValue(); | ||
| // really set up | ||
| OnlyHeaderOption headerOnly = Options.getOption(OnlyHeaderOption.class, opts); | ||
| initialize(filename, file, start, len, conf, headerOnly != null); | ||
| } catch (RemoteException re) { | ||
| log.error("Failed creating a WAL Reader: " + re.getMessage()); | ||
| if (re.getClassName().equals(WALConstants.LEASE_EXCEPTION_CLASS_NAME)) { | ||
| if (fs != null) { | ||
| fs.close(); | ||
| } | ||
| } | ||
| throw re; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why is this one upgraded from local variable to member field variable? The behavior definitely changes on repeated calls on the same object. Notice that this variable is mutated iteratively as part of the
whileloop inacquiredLease. Even if it happens and there's only one call ofacquireLeaseperFSWALobject I think it's a better practice to keep it local if that's what we intend to do here.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good catch!