chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed mq spi package

Browse Source
This commit is contained in:
agapple
2020-02-28 11:32:30 +08:00
parent 265a8e8594
commit ab74b2c82c
3 changed files with 46 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
connector/core/src/main/java/com/alibaba/otter/canal/connector/core/producer/MQDestination.java
View File

@@ -7,6 +7,7 @@ package com.alibaba.otter.canal.connector.core.producer;
* @version 1.0.0 * @version 1.0.0
*/ */
public class MQDestination { public class MQDestination {
private String canalDestination; private String canalDestination;
private String topic; private String topic;
private Integer partition; private Integer partition;

54
connector/core/src/main/java/com/alibaba/otter/canal/connector/core/spi/ExtensionLoader.java
View File

@@ -29,8 +29,7 @@ import org.slf4j.LoggerFactory;
*/ */
public class ExtensionLoader<T> { public class ExtensionLoader<T> {
private static final Logger logger = LoggerFactory private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class);
.getLogger(ExtensionLoader.class);
private static final String SERVICES_DIRECTORY = "META-INF/services/"; private static final String SERVICES_DIRECTORY = "META-INF/services/";
@@ -38,8 +37,7 @@ public class ExtensionLoader<T> {
private static final String DEFAULT_CLASSLOADER_POLICY = "internal"; private static final String DEFAULT_CLASSLOADER_POLICY = "internal";
private static final Pattern NAME_SEPARATOR = Pattern private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*");
.compile("\\s*[,]+\\s*");
private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>(); private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>();
@@ -180,8 +178,7 @@ public class ExtensionLoader<T> {
return instance; return instance;
} catch (Throwable t) { } catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+ ") could not be instantiated: " + t.getMessage(), + ") could not be instantiated: " + t.getMessage(), t);
t);
} }
} }
@@ -201,8 +198,7 @@ public class ExtensionLoader<T> {
return instance; return instance;
} catch (Throwable t) { } catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+ ") could not be instantiated: " + t.getMessage(), + ") could not be instantiated: " + t.getMessage(), t);
t);
} }
} }
@@ -268,8 +264,10 @@ public class ExtensionLoader<T> {
Map<String, Class<?>> extensionClasses = new HashMap<>(); Map<String, Class<?>> extensionClasses = new HashMap<>();
if (spiDir != null && standbyDir != null) { if (spiDir != null && standbyDir != null) {
// 1. plugin folder，customized extension classLoader （jar_dir/plugin） // 1. plugin folder，customized extension classLoader
String dir = File.separator + this.getJarDirectoryPath() + spiDir; // + "plugin"; // （jar_dir/plugin）
String dir = File.separator + this.getJarDirectoryPath() + spiDir; // +
// "plugin";
File externalLibDir = new File(dir); File externalLibDir = new File(dir);
if (!externalLibDir.exists()) { if (!externalLibDir.exists()) {
@@ -326,8 +324,7 @@ public class ExtensionLoader<T> {
try { try {
BufferedReader reader = null; BufferedReader reader = null;
try { try {
reader = new BufferedReader( reader = new BufferedReader(new InputStreamReader(url.openStream(), StandardCharsets.UTF_8));
new InputStreamReader(url.openStream(), StandardCharsets.UTF_8));
String line = null; String line = null;
while ((line = reader.readLine()) != null) { while ((line = reader.readLine()) != null) {
final int ci = line.indexOf('#'); final int ci = line.indexOf('#');
@@ -347,10 +344,12 @@ public class ExtensionLoader<T> {
// Class.forName(line, true, // Class.forName(line, true,
// classLoader); // classLoader);
if (!type.isAssignableFrom(clazz)) { if (!type.isAssignableFrom(clazz)) {
throw new IllegalStateException( throw new IllegalStateException("Error when load extension class(interface: "
"Error when load extension class(interface: " + type + type
+ ", class line: " + clazz.getName() + ", class line: "
+ "), class " + clazz.getName() + clazz.getName()
+ "), class "
+ clazz.getName()
+ "is not subtype of interface."); + "is not subtype of interface.");
} else { } else {
try { try {
@@ -368,9 +367,9 @@ public class ExtensionLoader<T> {
extensionClasses.put(n, clazz); extensionClasses.put(n, clazz);
} else if (c != clazz) { } else if (c != clazz) {
cachedNames.remove(clazz); cachedNames.remove(clazz);
throw new IllegalStateException( throw new IllegalStateException("Duplicate extension "
"Duplicate extension " + type.getName() + " name " + type.getName()
+ n + " on " + " name " + n + " on "
+ c.getName() + " and " + c.getName() + " and "
+ clazz.getName()); + clazz.getName());
} }
@@ -380,9 +379,12 @@ public class ExtensionLoader<T> {
} }
} }
} catch (Throwable t) { } catch (Throwable t) {
IllegalStateException e = new IllegalStateException( IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: "
"Failed to load extension class(interface: " + type + ", class line: " + type
+ line + ") in " + url + ", class line: "
+ line
+ ") in "
+ url
+ ", cause: " + ", cause: "
+ t.getMessage(), + t.getMessage(),
t); t);
@@ -397,15 +399,13 @@ public class ExtensionLoader<T> {
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + type + ", class file: " + url logger.error("Exception when load extension class(interface: " + type + ", class file: " + url
+ ") in " + url, + ") in " + url, t);
t);
} }
} // end of while urls } // end of while urls
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.error( logger.error("Exception when load extension class(interface: " + type + ", description file: " + fileName
"Exception when load extension class(interface: " + type + ", description file: " + fileName + ").", + ").", t);
t);
} }
} }

37
deployer/pom.xml
View File

@@ -84,7 +84,24 @@
</excludes> </excludes>
</configuration> </configuration>
</plugin> </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<version>2.10</version>
<executions>
<execution>
<id>copy-dependencies-to-canal-deployer</id>
<phase>package</phase>
<goals>
<goal>copy-dependencies</goal>
</goals>
<configuration>
<includeClassifiers>jar-with-dependencies</includeClassifiers>
<outputDirectory>${project.basedir}/target/canal/plugin</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId> <artifactId>maven-assembly-plugin</artifactId>
@@ -104,24 +121,6 @@
<attach>false</attach> <attach>false</attach>
</configuration> </configuration>
</plugin> </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-dependency-plugin</artifactId>
<version>2.10</version>
<executions>
<execution>
<id>copy-dependencies-to-canal-client-service</id>
<phase>package</phase>
<goals>
<goal>copy-dependencies</goal>
</goals>
<configuration>
<includeClassifiers>jar-with-dependencies</includeClassifiers>
<outputDirectory>${project.basedir}/target/canal/plugin</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>
</plugins> </plugins>
</build> </build>