Java SourceTaskContext类代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中org.apache.kafka.connect.source.SourceTaskContext类的典型用法代码示例。如果您正苦于以下问题：Java SourceTaskContext类的具体用法？Java SourceTaskContext怎么用？Java SourceTaskContext使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

SourceTaskContext类属于org.apache.kafka.connect.source包，在下文中一共展示了SourceTaskContext类的17个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: before

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@BeforeEach
public void before() {
  this.sourceTaskContext = mock(SourceTaskContext.class);
  this.offsetStorageReader = mock(OffsetStorageReader.class);
  when(this.sourceTaskContext.offsetStorageReader()).thenReturn(this.offsetStorageReader);
  this.task = new KinesisSourceTask();
  this.task.initialize(this.sourceTaskContext);
  this.kinesisClient = mock(AmazonKinesis.class);
  this.task.time = mock(Time.class);
  this.task.kinesisClientFactory = mock(KinesisClientFactory.class);
  when(this.task.kinesisClientFactory.create(any())).thenReturn(this.kinesisClient);

  this.settings = TestData.settings();
  this.config = new KinesisSourceConnectorConfig(this.settings);

}

开发者ID:jcustenborder，项目名称:kafka-connect-kinesis，代码行数:17，代码来源:KinesisSourceTaskTest.java

示例2: setup

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Before
public void setup() throws IOException, SQLException {
    String mysqlHost = "10.100.172.86";
    connection = DriverManager.getConnection("jdbc:mysql://" + mysqlHost + ":3306/mysql", "root", "passwd");
    
    config = new HashMap<>();
    config.put(MySqlSourceConnector.USER_CONFIG, "maxwell");
    config.put(MySqlSourceConnector.PASSWORD_CONFIG, "XXXXXX");
    config.put(MySqlSourceConnector.PORT_CONFIG, "3306");
    config.put(MySqlSourceConnector.HOST_CONFIG, mysqlHost);
    
    task = new MySqlSourceTask();
    offsetStorageReader = PowerMock.createMock(OffsetStorageReader.class);
    context = PowerMock.createMock(SourceTaskContext.class);
    task.initialize(context);

    runSql("drop table if exists test.users");
    runSql("drop database if exists test");
}

开发者ID:wushujames，项目名称:kafka-mysql-connector，代码行数:20，代码来源:MySqlSourceTaskTest.java

示例3: setup

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Before
public void setup() throws IOException {
    tempFile = File.createTempFile("file-stream-source-task-test", null);
    config = new HashMap<>();
    config.put(FileStreamSourceConnector.FILE_CONFIG, tempFile.getAbsolutePath());
    config.put(FileStreamSourceConnector.TOPIC_CONFIG, TOPIC);
    task = new FileStreamSourceTask();
    offsetStorageReader = PowerMock.createMock(OffsetStorageReader.class);
    context = PowerMock.createMock(SourceTaskContext.class);
    task.initialize(context);
}

开发者ID:wngn123，项目名称:wngn-jms-kafka，代码行数:12，代码来源:FileStreamSourceTaskTest.java

示例4: testPollsInBackground

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testPollsInBackground() throws Exception {
    createWorkerTask();

    sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
    EasyMock.expectLastCall();
    sourceTask.start(TASK_PROPS);
    EasyMock.expectLastCall();
    statusListener.onStartup(taskId);
    EasyMock.expectLastCall();

    final CountDownLatch pollLatch = expectPolls(10);
    // In this test, we don't flush, so nothing goes any further than the offset writer

    sourceTask.stop();
    EasyMock.expectLastCall();
    expectOffsetFlush(true);

    statusListener.onShutdown(taskId);
    EasyMock.expectLastCall();

    producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
    EasyMock.expectLastCall();

    transformationChain.close();
    EasyMock.expectLastCall();

    PowerMock.replayAll();

    workerTask.initialize(TASK_CONFIG);
    Future<?> taskFuture = executor.submit(workerTask);

    assertTrue(awaitLatch(pollLatch));
    workerTask.stop();
    assertTrue(workerTask.awaitStop(1000));

    taskFuture.get();

    PowerMock.verifyAll();
}

开发者ID:YMCoding，项目名称:kafka-0.11.0.0-src-with-comment，代码行数:41，代码来源:WorkerSourceTaskTest.java

示例5: initTask

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Before
public void initTask() {
    task = new FsSourceTask();
    taskConfig = new HashMap<String, String>() {{
        String uris[] = directories.stream().map(dir -> dir.toString())
                .toArray(size -> new String[size]);
        put(FsSourceTaskConfig.FS_URIS, String.join(",", uris));
        put(FsSourceTaskConfig.TOPIC, "topic_test");
        put(FsSourceTaskConfig.POLICY_CLASS, SimplePolicy.class.getName());
        put(FsSourceTaskConfig.FILE_READER_CLASS, TextFileReader.class.getName());
        put(FsSourceTaskConfig.POLICY_REGEXP, "^[0-9]*\\.txt$");
    }};

    //Mock initialization
    taskContext = PowerMock.createMock(SourceTaskContext.class);
    offsetStorageReader = PowerMock.createMock(OffsetStorageReader.class);

    EasyMock.expect(taskContext.offsetStorageReader())
            .andReturn(offsetStorageReader);

    EasyMock.expect(taskContext.offsetStorageReader())
            .andReturn(offsetStorageReader);

    EasyMock.expect(offsetStorageReader.offset(EasyMock.anyObject()))
            .andReturn(new HashMap<String, Object>() {{
                put("offset", 5L);
            }});
    EasyMock.expect(offsetStorageReader.offset(EasyMock.anyObject()))
            .andReturn(new HashMap<String, Object>() {{
                put("offset", 5L);
            }});

    EasyMock.checkOrder(taskContext, false);
    EasyMock.replay(taskContext);

    EasyMock.checkOrder(offsetStorageReader, false);
    EasyMock.replay(offsetStorageReader);

    task.initialize(taskContext);

}

开发者ID:mmolimar，项目名称:kafka-connect-fs，代码行数:42，代码来源:FsSourceTaskTestBase.java

示例6: testPause

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testPause() throws Exception {
    createWorkerTask();

    sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
    EasyMock.expectLastCall();
    sourceTask.start(TASK_PROPS);
    EasyMock.expectLastCall();
    statusListener.onStartup(taskId);
    EasyMock.expectLastCall();

    AtomicInteger count = new AtomicInteger(0);
    CountDownLatch pollLatch = expectPolls(10, count);
    // In this test, we don't flush, so nothing goes any further than the offset writer

    statusListener.onPause(taskId);
    EasyMock.expectLastCall();

    sourceTask.stop();
    EasyMock.expectLastCall();
    expectOffsetFlush(true);

    statusListener.onShutdown(taskId);
    EasyMock.expectLastCall();

    producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
    EasyMock.expectLastCall();

    transformationChain.close();
    EasyMock.expectLastCall();

    PowerMock.replayAll();

    workerTask.initialize(TASK_CONFIG);
    Future<?> taskFuture = executor.submit(workerTask);
    assertTrue(awaitLatch(pollLatch));

    workerTask.transitionTo(TargetState.PAUSED);

    int priorCount = count.get();
    Thread.sleep(100);

    // since the transition is observed asynchronously, the count could be off by one loop iteration
    assertTrue(count.get() - priorCount <= 1);

    workerTask.stop();
    assertTrue(workerTask.awaitStop(1000));

    taskFuture.get();

    PowerMock.verifyAll();
}

开发者ID:YMCoding，项目名称:kafka-0.11.0.0-src-with-comment，代码行数:53，代码来源:WorkerSourceTaskTest.java

示例7: testFailureInPoll

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testFailureInPoll() throws Exception {
    createWorkerTask();

    sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
    EasyMock.expectLastCall();
    sourceTask.start(TASK_PROPS);
    EasyMock.expectLastCall();
    statusListener.onStartup(taskId);
    EasyMock.expectLastCall();

    final CountDownLatch pollLatch = new CountDownLatch(1);
    final RuntimeException exception = new RuntimeException();
    EasyMock.expect(sourceTask.poll()).andAnswer(new IAnswer<List<SourceRecord>>() {
        @Override
        public List<SourceRecord> answer() throws Throwable {
            pollLatch.countDown();
            throw exception;
        }
    });

    statusListener.onFailure(taskId, exception);
    EasyMock.expectLastCall();

    sourceTask.stop();
    EasyMock.expectLastCall();
    expectOffsetFlush(true);

    producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
    EasyMock.expectLastCall();

    transformationChain.close();
    EasyMock.expectLastCall();

    PowerMock.replayAll();

    workerTask.initialize(TASK_CONFIG);
    Future<?> taskFuture = executor.submit(workerTask);

    assertTrue(awaitLatch(pollLatch));
    workerTask.stop();
    assertTrue(workerTask.awaitStop(1000));

    taskFuture.get();

    PowerMock.verifyAll();
}

开发者ID:YMCoding，项目名称:kafka-0.11.0.0-src-with-comment，代码行数:48，代码来源:WorkerSourceTaskTest.java

示例8: testCommit

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testCommit() throws Exception {
    // Test that the task commits properly when prompted
    createWorkerTask();

    sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
    EasyMock.expectLastCall();
    sourceTask.start(TASK_PROPS);
    EasyMock.expectLastCall();
    statusListener.onStartup(taskId);
    EasyMock.expectLastCall();

    // We'll wait for some data, then trigger a flush
    final CountDownLatch pollLatch = expectPolls(1);
    expectOffsetFlush(true);

    sourceTask.stop();
    EasyMock.expectLastCall();
    expectOffsetFlush(true);

    statusListener.onShutdown(taskId);
    EasyMock.expectLastCall();

    producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
    EasyMock.expectLastCall();

    transformationChain.close();
    EasyMock.expectLastCall();

    PowerMock.replayAll();

    workerTask.initialize(TASK_CONFIG);
    Future<?> taskFuture = executor.submit(workerTask);

    assertTrue(awaitLatch(pollLatch));
    assertTrue(workerTask.commitOffsets());
    workerTask.stop();
    assertTrue(workerTask.awaitStop(1000));

    taskFuture.get();

    PowerMock.verifyAll();
}

开发者ID:YMCoding，项目名称:kafka-0.11.0.0-src-with-comment，代码行数:44，代码来源:WorkerSourceTaskTest.java

示例9: testCommitFailure

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testCommitFailure() throws Exception {
    // Test that the task commits properly when prompted
    createWorkerTask();

    sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
    EasyMock.expectLastCall();
    sourceTask.start(TASK_PROPS);
    EasyMock.expectLastCall();
    statusListener.onStartup(taskId);
    EasyMock.expectLastCall();

    // We'll wait for some data, then trigger a flush
    final CountDownLatch pollLatch = expectPolls(1);
    expectOffsetFlush(true);

    sourceTask.stop();
    EasyMock.expectLastCall();
    expectOffsetFlush(false);

    statusListener.onShutdown(taskId);
    EasyMock.expectLastCall();

    producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
    EasyMock.expectLastCall();

    transformationChain.close();
    EasyMock.expectLastCall();

    PowerMock.replayAll();

    workerTask.initialize(TASK_CONFIG);
    Future<?> taskFuture = executor.submit(workerTask);

    assertTrue(awaitLatch(pollLatch));
    assertTrue(workerTask.commitOffsets());
    workerTask.stop();
    assertTrue(workerTask.awaitStop(1000));

    taskFuture.get();

    PowerMock.verifyAll();
}

开发者ID:YMCoding，项目名称:kafka-0.11.0.0-src-with-comment，代码行数:44，代码来源:WorkerSourceTaskTest.java

示例10: testSlowTaskStart

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testSlowTaskStart() throws Exception {
    final CountDownLatch startupLatch = new CountDownLatch(1);
    final CountDownLatch finishStartupLatch = new CountDownLatch(1);

    createWorkerTask();

    sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
    EasyMock.expectLastCall();
    sourceTask.start(TASK_PROPS);
    EasyMock.expectLastCall().andAnswer(new IAnswer<Object>() {
        @Override
        public Object answer() throws Throwable {
            startupLatch.countDown();
            assertTrue(awaitLatch(finishStartupLatch));
            return null;
        }
    });

    statusListener.onStartup(taskId);
    EasyMock.expectLastCall();

    sourceTask.stop();
    EasyMock.expectLastCall();
    expectOffsetFlush(true);

    statusListener.onShutdown(taskId);
    EasyMock.expectLastCall();

    producer.close(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class));
    EasyMock.expectLastCall();

    transformationChain.close();
    EasyMock.expectLastCall();

    PowerMock.replayAll();

    workerTask.initialize(TASK_CONFIG);
    Future<?> workerTaskFuture = executor.submit(workerTask);

    // Stopping immediately while the other thread has work to do should result in no polling, no offset commits,
    // exiting the work thread immediately, and the stop() method will be invoked in the background thread since it
    // cannot be invoked immediately in the thread trying to stop the task.
    assertTrue(awaitLatch(startupLatch));
    workerTask.stop();
    finishStartupLatch.countDown();
    assertTrue(workerTask.awaitStop(1000));

    workerTaskFuture.get();

    PowerMock.verifyAll();
}

开发者ID:YMCoding，项目名称:kafka-0.11.0.0-src-with-comment，代码行数:53，代码来源:WorkerSourceTaskTest.java

示例11: poll

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
protected void poll(final String packageName, TestCase testCase) throws InterruptedException, IOException {
  String keySchemaConfig = ObjectMapperFactory.INSTANCE.writeValueAsString(testCase.keySchema);
  String valueSchemaConfig = ObjectMapperFactory.INSTANCE.writeValueAsString(testCase.valueSchema);

  Map<String, String> settings = Maps.newLinkedHashMap();
  settings.put(SpoolDirSourceConnectorConfig.INPUT_PATH_CONFIG, this.inputPath.getAbsolutePath());
  settings.put(SpoolDirSourceConnectorConfig.FINISHED_PATH_CONFIG, this.finishedPath.getAbsolutePath());
  settings.put(SpoolDirSourceConnectorConfig.ERROR_PATH_CONFIG, this.errorPath.getAbsolutePath());
  settings.put(SpoolDirSourceConnectorConfig.INPUT_FILE_PATTERN_CONF, String.format("^.*\\.%s", packageName));
  settings.put(SpoolDirSourceConnectorConfig.TOPIC_CONF, "testing");
  settings.put(SpoolDirSourceConnectorConfig.KEY_SCHEMA_CONF, keySchemaConfig);
  settings.put(SpoolDirSourceConnectorConfig.VALUE_SCHEMA_CONF, valueSchemaConfig);
  settings.put(SpoolDirSourceConnectorConfig.EMPTY_POLL_WAIT_MS_CONF, "10");
  settings(settings);
  if (null != testCase.settings && !testCase.settings.isEmpty()) {
    settings.putAll(testCase.settings);
  }

  this.task = createTask();

  SourceTaskContext sourceTaskContext = mock(SourceTaskContext.class);
  OffsetStorageReader offsetStorageReader = mock(OffsetStorageReader.class);
  when(offsetStorageReader.offset(anyMap())).thenReturn(testCase.offset);
  when(sourceTaskContext.offsetStorageReader()).thenReturn(offsetStorageReader);
  this.task.initialize(sourceTaskContext);

  this.task.start(settings);

  String dataFile = new File(packageName, Files.getNameWithoutExtension(testCase.path.toString())) + ".data";
  log.trace("poll(String, TestCase) - dataFile={}", dataFile);

  String inputFileName = String.format("%s.%s",
      Files.getNameWithoutExtension(testCase.path.toString()),
      packageName
  );


  final File inputFile = new File(this.inputPath, inputFileName);
  log.trace("poll(String, TestCase) - inputFile = {}", inputFile);
  final File processingFile = this.task.processingFile(inputFile);
  try (InputStream inputStream = this.getClass().getResourceAsStream(dataFile)) {
    try (OutputStream outputStream = new FileOutputStream(inputFile)) {
      ByteStreams.copy(inputStream, outputStream);
    }
  }

  assertFalse(processingFile.exists(), String.format("processingFile %s should not exist before first poll().", processingFile));
  assertTrue(inputFile.exists(), String.format("inputFile %s should exist.", inputFile));
  List<SourceRecord> records = this.task.poll();
  assertTrue(inputFile.exists(), String.format("inputFile %s should exist after first poll().", inputFile));
  assertTrue(processingFile.exists(), String.format("processingFile %s should exist after first poll().", processingFile));

  assertNotNull(records, "records should not be null.");
  assertFalse(records.isEmpty(), "records should not be empty");
  assertEquals(testCase.expected.size(), records.size(), "records.size() does not match.");

  for (int i = 0; i < testCase.expected.size(); i++) {
    SourceRecord expectedRecord = testCase.expected.get(i);
    SourceRecord actualRecord = records.get(i);
    assertSourceRecord(expectedRecord, actualRecord, String.format("index:%s", i));
  }

  records = this.task.poll();
  assertTrue(records.isEmpty(), "records should be null after first poll.");
  records = this.task.poll();
  assertTrue(records.isEmpty(), "records should be null after first poll.");
  assertFalse(inputFile.exists(), String.format("inputFile %s should not exist.", inputFile));
  assertFalse(processingFile.exists(), String.format("processingFile %s should not exist.", processingFile));

  assertTrue(records.isEmpty(), "records should be empty.");

  final File finishedFile = new File(this.finishedPath, inputFileName);
  assertTrue(finishedFile.exists(), String.format("finishedFile %s should exist.", finishedFile));

}

开发者ID:jcustenborder，项目名称:kafka-connect-spooldir，代码行数:76，代码来源:SpoolDirSourceTaskTest.java

示例12: initialize

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Override
public void initialize(SourceTaskContext context) {

	super.initialize(context);
	LOG.info("AMQP source task initialized");
}

开发者ID:ppatierno，项目名称:kafka-connect-amqp，代码行数:7，代码来源:AmqpSourceTask.java

示例13: setUp

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Override
    public void setUp() {
        offsets = new HashMap<>();
        totalWrittenDocuments = 0;
        try {
            super.setUp();
            mongodStarter = MongodStarter.getDefaultInstance();
            mongodConfig = new MongodConfigBuilder()
                    .version(Version.Main.V3_2)
                    .replication(new Storage(REPLICATION_PATH, "rs0", 1024))
                    .net(new Net(12345, Network.localhostIsIPv6()))
                    .build();
            mongodExecutable = mongodStarter.prepare(mongodConfig);
            mongod = mongodExecutable.start();
            mongoClient = new MongoClient(new ServerAddress("localhost", 12345));
            MongoDatabase adminDatabase = mongoClient.getDatabase("admin");

            BasicDBObject replicaSetSetting = new BasicDBObject();
            replicaSetSetting.put("_id", "rs0");
            BasicDBList members = new BasicDBList();
            DBObject host = new BasicDBObject();
            host.put("_id", 0);
            host.put("host", "127.0.0.1:12345");
            members.add(host);
            replicaSetSetting.put("members", members);
            adminDatabase.runCommand(new BasicDBObject("isMaster", 1));
            adminDatabase.runCommand(new BasicDBObject("replSetInitiate", replicaSetSetting));
            MongoDatabase db = mongoClient.getDatabase("mydb");
            db.createCollection("test1");
            db.createCollection("test2");
            db.createCollection("test3");
        } catch (Exception e) {
//                Assert.assertTrue(false);
        }

        task = new MongodbSourceTask();

        offsetStorageReader = PowerMock.createMock(OffsetStorageReader.class);
        context = PowerMock.createMock(SourceTaskContext.class);
        task.initialize(context);

        sourceProperties = new HashMap<>();
        sourceProperties.put("uri", "mongodb://localhost:12345");
        sourceProperties.put("batch.size", Integer.toString(100));
        sourceProperties.put("schema.name", "schema");
        sourceProperties.put("topic.prefix", "prefix");
        sourceProperties.put("databases", "mydb.test1,mydb.test2,mydb.test3");

    }

开发者ID:DataReply，项目名称:kafka-connect-mongodb，代码行数:50，代码来源:MongodbSourceUriTaskTest.java

示例14: setUp

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Override
    public void setUp() {
        offsets = new HashMap<>();
        totalWrittenDocuments = 0;
        try {
            super.setUp();
            mongodStarter = MongodStarter.getDefaultInstance();
            mongodConfig = new MongodConfigBuilder()
                    .version(Version.Main.V3_2)
                    .replication(new Storage(REPLICATION_PATH, "rs0", 1024))
                    .net(new Net(12345, Network.localhostIsIPv6()))
                    .build();
            mongodExecutable = mongodStarter.prepare(mongodConfig);
            mongod = mongodExecutable.start();
            mongoClient = new MongoClient(new ServerAddress("localhost", 12345));
            MongoDatabase adminDatabase = mongoClient.getDatabase("admin");

            BasicDBObject replicaSetSetting = new BasicDBObject();
            replicaSetSetting.put("_id", "rs0");
            BasicDBList members = new BasicDBList();
            DBObject host = new BasicDBObject();
            host.put("_id", 0);
            host.put("host", "127.0.0.1:12345");
            members.add(host);
            replicaSetSetting.put("members", members);
            adminDatabase.runCommand(new BasicDBObject("isMaster", 1));
            adminDatabase.runCommand(new BasicDBObject("replSetInitiate", replicaSetSetting));
            MongoDatabase db = mongoClient.getDatabase("mydb");
            db.createCollection("test1");
            db.createCollection("test2");
            db.createCollection("test3");
        } catch (Exception e) {
//                Assert.assertTrue(false);
        }

        task = new MongodbSourceTask();

        offsetStorageReader = PowerMock.createMock(OffsetStorageReader.class);
        context = PowerMock.createMock(SourceTaskContext.class);
        task.initialize(context);

        sourceProperties = new HashMap<>();
        sourceProperties.put("host", "localhost");
        sourceProperties.put("port", Integer.toString(12345));
        sourceProperties.put("batch.size", Integer.toString(100));
        sourceProperties.put("schema.name", "schema");
        sourceProperties.put("topic.prefix", "prefix");
        sourceProperties.put("databases", "mydb.test1,mydb.test2,mydb.test3");

    }

开发者ID:DataReply，项目名称:kafka-connect-mongodb，代码行数:51，代码来源:MongodbSourceTaskTest.java

示例15: initialize

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Override
public void initialize(SourceTaskContext context) {
    super.initialize(context);
}

开发者ID:yaravind，项目名称:kafka-connect-jenkins，代码行数:5，代码来源:JenkinsSourceTask.java

示例16: testSourceTaskProcessAll

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testSourceTaskProcessAll() {
    logger.info (
        "Setting up stubs for clean run with no published messages"
    );
    
    SourceTaskContext context = EasyMock.createMock(
        SourceTaskContext.class
    );
    
    OffsetStorageReader offsetStorageReader = EasyMock.createMock (
        OffsetStorageReader.class
    );
    
    EasyMock.expect(context.offsetStorageReader())
        .andReturn(offsetStorageReader);
    
    EasyMock.expect(offsetStorageReader.offsets(offsetPartitions()))
        .andReturn(new HashMap<Map<String, String>, Map<String, Object>>());
    
    EasyMock.checkOrder(context, false);
    EasyMock.replay(context);
    
    EasyMock.checkOrder(offsetStorageReader, false);
    EasyMock.replay(offsetStorageReader);
    
    logger.info ("Source task poll all - test");
    
    ReplicateSourceTask sourceTask = new ReplicateSourceTask();
    
    sourceTask.initialize(context);
    
    try {
        sourceTask.start(getConfigPropsForSourceTask());
        
        assertEquals (
            "Expecting no record for first poll, nothing in PLOG", 
            null, 
            sourceTask.poll()
        );
        
        /* expect 2 messages per PLOG, 3 PLOGs with data */
        List<SourceRecord> records = sourceTask.poll();
        validateRecordCount (records.size());
        validateTopics (records);
        
        records = sourceTask.poll();
        validateRecordCount (records.size());
        validateTopics (records);
        
        records = sourceTask.poll();
        validateRecordCount (records.size());
        validateTopics (records);
    }
    catch (InterruptedException ie) {
        logger.info ("Stopping task");
    }
    catch (Exception e) {
        e.printStackTrace();
        fail (e.getMessage());
    }
    finally {
        sourceTask.stop();
    }
}

开发者ID:dbvisitsoftware，项目名称:replicate-connector-for-kafka，代码行数:66，代码来源:ReplicateSourceTaskTest.java

示例17: testSourceTaskProcessTwoMessagesBySCN

import org.apache.kafka.connect.source.SourceTaskContext; //导入依赖的package包/类
@Test
public void testSourceTaskProcessTwoMessagesBySCN() {
    logger.info (
        "Setting up stubs using global SCN filter for fake cold start"
    );
    
    SourceTaskContext context = EasyMock.createMock(
        SourceTaskContext.class
    );
    
    OffsetStorageReader offsetStorageReader = EasyMock.createMock (
        OffsetStorageReader.class
    );
    
    EasyMock.expect(context.offsetStorageReader())
        .andReturn(offsetStorageReader);
    
    EasyMock.expect(offsetStorageReader.offsets(offsetPartitions()))
        .andReturn(new HashMap<Map<String, String>, Map<String, Object>>());
    
    EasyMock.checkOrder(context, false);
    EasyMock.replay(context);
    
    EasyMock.checkOrder(offsetStorageReader, false);
    EasyMock.replay(offsetStorageReader);
    
    logger.info ("Source task poll all committed, no new data");
    
    ReplicateSourceTask sourceTask = new ReplicateSourceTask();
    
    sourceTask.initialize(context);
    
    sourceTask.start(
        getConfigPropsForSourceTaskWithStartSCN(PLOG_22_TRANSACTION_SCN)
    );
    
    try {
        /* expect no new messages to publish for first poll */
        List<SourceRecord> records = sourceTask.poll();
        assertNull (
            "Expecting no records for first PLOG 20, contains no data",
            records
        );
        
        records = sourceTask.poll();
        assertNull (
            "Expecting no records for second PLOG",
            records
        );
        
        /* expect 2 messages for last two PLOG */
        records = sourceTask.poll();
        validateRecordCount (records.size());
        validateTopics (records);
        validateRecordOffset(records, PLOG_21_TRANSACTION_OFFSET);
        
        records = sourceTask.poll();
        validateRecordCount (records.size());
        validateTopics (records);
        validateRecordOffset(records, PLOG_22_TRANSACTION_OFFSET);
        
        /* next call will block because there is no further data */
    }
    catch (InterruptedException ie) {
        logger.info ("Stopping task");
    }
    catch (Exception e) {
        e.printStackTrace();
        fail (e.getMessage());
    }
    finally {
        sourceTask.stop();
    }
}

开发者ID:dbvisitsoftware，项目名称:replicate-connector-for-kafka，代码行数:75，代码来源:ReplicateSourceTaskTest.java

注：本文中的org.apache.kafka.connect.source.SourceTaskContext类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Java Whirlpool0类代码示例发布时间：2022-05-22

Java Tree类代码示例发布时间：2022-05-22

剪的笔顺,诠释剪的笔画,认识剪的部首

1 六六分期app的软件客服如何联系？(六六分期

六六分期app的软件客服如何联系？不知道吗？加qq群【895510560】即可！标题：六六分期

阅读：18090|2023-10-27

2 可心卡盟:win10系统火狐flash插件崩溃怎么

今天小编告诉大家如何处理win10系统火狐flash插件总是崩溃的问题，可能很多用户都不知

阅读：9617|2022-11-06

3 亲亲特价:怎么删除回收站图标

今天小编告诉大家如何对win10系统删除桌面回收站图标进行设置，可能很多用户都不知道

阅读：8149|2022-11-06

4 济南大学虚拟社区:鲁大师节能降温的具体办

今天小编告诉大家如何对win10系统电脑设置节能降温的设置方法，想必大家都遇到过需要

阅读：8530|2022-11-06

5 xlueops.exe:无线网络安装向导

我们在使用xp系统的过程中,经常需要对xp系统无线网络安装向导设置进行设置，可能很多

阅读：8432|2022-11-06

6 女斗合众国:win7系统cf与主机连接不稳定怎

今天小编告诉大家如何处理win7系统玩cf老是与主机连接不稳定的问题，可能很多用户都不

阅读：9346|2022-11-06

7 0xc000022-[cf烟雾头]cf怎么调烟雾头

电脑对日常生活的重要性小编就不多说了，可是一旦碰到win7系统设置cf烟雾头的问题，很

阅读：8397|2022-11-06

8 qizideyouhuo:应用程序无法正常启动0xc0000

我们在日常使用电脑的时候，有的小伙伴们可能在打开应用的时候会遇见提示应用程序无法

阅读：7833|2022-11-06

9 ipz-185:win7系统vcf文件怎么打开

今天小编告诉大家如何对win7系统打开vcf文件进行设置，可能很多用户都不知道怎么对win

阅读：8387|2022-11-06

10 傻哥蹦迪:win10系统s4怎么打开usb调试

今天小编告诉大家如何对win10系统s4开启USB调试模式进行设置，可能很多用户都不知道怎

阅读：7380|2022-11-06

客服电话

电子邮件

Java SourceTaskContext类代码示例

示例1: before

示例2: setup

示例3: setup

示例4: testPollsInBackground

示例5: initTask

示例6: testPause

示例7: testFailureInPoll

示例8: testCommit

示例9: testCommitFailure

示例10: testSlowTaskStart

示例11: poll

示例12: initialize

示例13: setUp

示例14: setUp

示例15: initialize

示例16: testSourceTaskProcessAll

示例17: testSourceTaskProcessTwoMessagesBySCN

请发表评论

全部评论

上一篇：

下一篇：

summerblue/laravel-shop: Laravel 电商实

krishnaik06/Machine-Learning-in-90-days

如何用微信小程序，每天给自己赚个鸡腿？

armancodv/building-energy-model-matlab:

美元符号为什么是“$”

剪的笔顺,诠释剪的笔画,认识剪的部首

六六分期app的软件客服如何联系？(六六分期

florent37/ViewAnimator: A fluent Android

florent37/Shrine-MaterialDesign2: implem

CVE-2020-36276

SimpleSoftwareIO/simple-sms: Send and re

关于我们

产品与服务

解决方案

139-2527-9053