我如何关闭Apache Kafka连接器任务?

问题描述 投票:1回答:2

现在我正在使用Apache Kafka并且有任务:我们在目录中有一些csv文件,它是一个小批量文件,每个文件大约是25-30 MB。我只需要 - 解析文件并将其放入kafka。

正如我所看到的,Kafka有一些像Connector这样有趣的东西。

我可以创建Source-Connector和SourceTask,但我不明白一件事:当我处理文件时,我如何停止或删除我的任务?

例如,我有虚拟连接器:

public class DummySourceConnector extends SourceConnector {
private static final Logger logger = LogManager.getLogger();

@Override
public String version() {
    logger.info("version");

    return "1";
}

@Override
public ConfigDef config() {
    logger.info("config");

    return null;
}

@Override
public Class<? extends Task> taskClass() {
    return DummySourceTask.class;
}

@Override
public void start(Map<String, String> props) {
    logger.info("start {}", props);
}

@Override
public void stop() {
    logger.info("stop");
}

@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
    logger.info("taskConfigs {}", maxTasks);

    return ImmutableList.of(ImmutableMap.of("key", "value"));
}

任务:

public class DummySourceTask extends SourceTask {
private static final Logger logger = LogManager.getLogger();

private long offset = 0;

@Override
public String version() {
    logger.info("version");

    return "1";
}

@Override
public void start(Map<String, String> props) {
    logger.info("start {}", props);
}


@Override
public List<SourceRecord> poll() throws InterruptedException {
    Thread.sleep(3000);

    final String value = "Offset " + offset++ + " Timestamp " + Instant.now().toString();

    logger.info("poll value {}", value);

    return ImmutableList.of(new SourceRecord(
            ImmutableMap.of("partition", 0),
            ImmutableMap.of("offset", offset),
            "topic-dummy",
            SchemaBuilder.STRING_SCHEMA,
            value
    ));
}

public void stop() {
    logger.info("stop");
}

但是,当这一切完成后,我怎么能完成任务呢?或者也许你可以帮助我完成这个任务的另一个想法。

谢谢你的帮助!

java apache-kafka apache-kafka-connect
2个回答
1
投票

首先,我建议您查看现有的连接器here。我觉得spooldir连接器会对你有所帮助。您甚至可以只需下载并安装它,而无需编写任何代码。

其次,如果我理解正确,你想要停止任务。我相信this discussion是你想要的。


0
投票

在事件发生时终止Task的不太优雅的解决方案是检查任务源中的事件并调用System.exit(1)。

然而,我发现的最优雅的解决方案是:

当事件发生时,连接器任务将REST调用应用于代理,以便停止运行任务的连接器。

要执行此操作,任务本身应该知道运行任务的连接器的名称,您可以按照此discussion的步骤找到该名称。

因此,连接器的名称在Task的属性参数中,存在一个带有“name”键的属性,其值是执行Task的Connector的名称(如果发生事件,我们想要停止)。

最后,我们进行REST调用,如果任务停止,我们得到204回答,没有内容。

通话代码如下:

 try {

  URL url = new URL("url/" + connectorName);
  HttpURLConnection conn = (HttpURLConnection) url.openConnection();
  conn.setRequestMethod("DELETE");
  conn.setRequestProperty("Accept", "application/json");

  if (conn.getResponseCode() != 204) {
    throw new RuntimeException("Failed : HTTP error code : "
        + conn.getResponseCode());
  }

  BufferedReader br = new BufferedReader(new InputStreamReader(
    (conn.getInputStream())));

  String output;
  System.out.println("Task Stopped \n");
  while ((output = br.readLine()) != null) {
    System.out.println(output);
  }

  conn.disconnect();

  } catch (MalformedURLException e) {

  e.printStackTrace();

  } catch (IOException e) {

  e.printStackTrace();

  }

现在所有连接器任务都停止了。

(当然,正如前面提到的,你必须记住每个SourceTask和每个SinkTask的逻辑都是无穷无尽的。如果事件发生,它们应该永远不会停止,而是连续搜索你提供它们的文件中的新条目所以通常你会通过REST调用来阻止它们,如果你希望它们在事件发生时停止,你可以将REST调用放在他们自己的代码中。)

© www.soinside.com 2019 - 2024. All rights reserved.