hadoop - Flume - 如何为 JMS Source (Tibco) 指定用户名/密码

Question

我最近一直在尝试使用 Flume。我目前正在使用 JMS 源 (http://flume.apache.org/FlumeUserGuide.html#jms-source)

但是，当我尝试运行水槽代理时，它成功创建了 channel 和接收器，但在创建源时失败了。我正在使用 TIBCO EMS，我需要使用 EMS 用户名/密码组合进行身份验证。

文档要求我定义一个密码文件，其中包含目标/提供者的密码。是否有创建此文件的正确方法(我目前将密码存储在文本文件中)。

这是错误堆栈(我很确定用户/密码是正确的)

13/11/17 11:49:08 INFO source.DefaultSourceFactory: Creating instance of source WeatherData, type jms
13/11/17 11:49:08 ERROR node.AbstractConfigurationProvider: Source WeatherData has been removed due to an error during configuration
org.apache.flume.FlumeException: Could not lookup ConnectionFactory
        at org.apache.flume.source.jms.JMSSource.doConfigure(JMSSource.java:222)
        at org.apache.flume.source.BasicSourceSemantics.configure(BasicSourceSemantics.java:65)
        at org.apache.flume.conf.Configurables.configure(Configurables.java:41)
        at org.apache.flume.node.AbstractConfigurationProvider.loadSources(AbstractConfigurationProvider.java:331)
        at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:102)
        at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441)
        at java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317)
        at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecutor.java:98)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.runPeriodic(ScheduledThreadPoolExecutor.java:180)
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:204)
        at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
        at java.lang.Thread.run(Thread.java:662)
Caused by: javax.naming.AuthenticationException: Not permitted: invalid name or password [Root exception is javax.jms.JMSSecurityException: invalid name or password]
        at com.tibco.tibjms.naming.TibjmsContext.lookup(TibjmsContext.java:668)
        at com.tibco.tibjms.naming.TibjmsContext.lookup(TibjmsContext.java:489)
        at javax.naming.InitialContext.lookup(InitialContext.java:392)
        at org.apache.flume.source.jms.JMSSource.doConfigure(JMSSource.java:219)
        ... 14 more
Caused by: javax.jms.JMSSecurityException: invalid name or password
        at com.tibco.tibjms.Tibjmsx.buildException(Tibjmsx.java:543)
        at com.tibco.tibjms.TibjmsConnection._create(TibjmsConnection.java:1044)
        at com.tibco.tibjms.TibjmsConnection.<init>(TibjmsConnection.java:2707)
        at com.tibco.tibjms.TibjmsQueueConnection.<init>(TibjmsQueueConnection.java:36)
        at com.tibco.tibjms.TibjmsxCFImpl._createImpl(TibjmsxCFImpl.java:186)
        at com.tibco.tibjms.TibjmsxCFImpl._createConnection(TibjmsxCFImpl.java:239)
        at com.tibco.tibjms.TibjmsQueueConnectionFactory.createQueueConnection(TibjmsQueueConnectionFactory.java:87)
        at com.tibco.tibjms.naming.TibjmsContext$Messenger.request(TibjmsContext.java:325)
        at com.tibco.tibjms.naming.TibjmsContext.lookup(TibjmsContext.java:655)
        ... 17 more

Answer 1

关于密码文件不确定，这似乎是 Flume 特定的。
TIBCO EMS 在 2 个级别上进行身份验证:

查找connectionFactory

目的地查找

一些“预构建”配置，如 Spring 等(可能是水槽)与此斗争
并且需要一些返工。

检查下面的代码，您可以在其中看到用户/密码应用于两次级别以获得
来自 TIBCO EMS 的队列 queue.sample:

/**
 * @param args
 */
public static void main(String[] args) {

    String userName = "user";
    String password = "password";
    try {
        // Obtain a JNDI connection
        Properties env = new Properties();
        env.put(Context.INITIAL_CONTEXT_FACTORY,
                "com.tibco.tibjms.naming.TibjmsInitialContextFactory");
        env.put(Context.PROVIDER_URL, "tibjmsnaming://localhost:7222");
        env.put(Context.SECURITY_PRINCIPAL, userName);
        env.put(Context.SECURITY_CREDENTIALS, password);
        // ... specify the JNDI properties specific to the vendor
        InitialContext jndi = new InitialContext(env);

        ConnectionFactory factory = (ConnectionFactory) jndi
                .lookup("ConnectionFactory");

        try {
            Connection connection = factory.createConnection(userName,
                    password);
            Session session = connection.createSession();

            Queue sampleTopic = (Queue) jndi.lookup("queue.sample");

        } catch (JMSException je) {
            je.printStackTrace();

        }

    } catch (NamingException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }

}